Use epoch-based expressions for identity date/timestamp partition pushdown

Identity transforms on date/timestamp types now produce epoch integers
in Hive-style partition paths instead of native text. This avoids DuckDB
text formatting issues with BC dates (e.g. "4713-01-01 (BC)").

- Date identity: datediff('day', date '1970-01-01', col::date) → epoch days
- Timestamp identity: epoch_us(col) → epoch microseconds
- New DeserializePartitionValueFromEpochInteger converts epoch integers
  back to Iceberg binary format during path parsing
- Added cross-path validation tests for identity date/timestamp
- Added infinity identity partition clamped test

Signed-off-by: Marco Slot <marco.slot@snowflake.com>

Author: sfc-gh-mslot

pg_lake_table/include/pg_lake/fdw/partition_transform.h

@@ -33,6 +33,8 @@ extern List *AllPartitionTransformList(Oid relationId);
 extern List *GetPartitionTransformsFromSpecFields(Oid relationId, List *specFields);
 extern void *DeserializePartitionValueFromPGText(IcebergPartitionTransform * transform,
 												 const char *valueText, size_t *valueLength);
+extern void *DeserializePartitionValueFromEpochInteger(IcebergPartitionTransform * transform,
+													   const char *epochText, size_t *valueLength);
 extern const char *SerializePartitionValueToPGText(void *value, size_t valueLength, IcebergPartitionTransform * transform);
 extern Datum PartitionValueToDatum(IcebergPartitionTransformType transformType, void *value, size_t valueLength,
 								   PGType pgType, bool *isNull);

pg_lake_table/src/fdw/partition_pushdown.c

@@ -25,15 +25,13 @@
 #include "postgres.h"
 
 #include "utils/builtins.h"
+#include "utils/lsyscache.h"
 
 #include "pg_lake/fdw/partition_pushdown.h"
 #include "pg_lake/fdw/partition_transform.h"
 #include "pg_lake/iceberg/api/partitioning.h"
 #include "pg_lake/iceberg/manifest_spec.h"
 
-/* prefix for synthetic partition columns in COPY TO queries */
-#define PARTITION_COLUMN_PREFIX "__part_"
-
 
 static char *PartitionTransformToDuckDBExpression(IcebergPartitionTransform * transform);
 
@@ -87,13 +85,21 @@ PartitionTransformToDuckDBExpression(IcebergPartitionTransform * transform)
 		case PARTITION_TRANSFORM_IDENTITY:
 			{
 				/*
-				 * Identity partitions use the column value directly.  DuckDB
-				 * writes the value in its default text format into the
-				 * Hive-style path (e.g. __part_0=2025-01-15), and
-				 * DeserializePartitionValueFromPGText parses it back using
-				 * the transform's result PG type.
+				 * Identity partitions use the column value directly for
+				 * non-temporal types. For date/timestamp types, we produce
+				 * epoch integers to avoid DuckDB text formatting issues (e.g.
+				 * BC dates formatted as "4713-01-01 (BC)").
+				 *
+				 * ParsePartitionValuesFromPath uses
+				 * DeserializePartitionValueFromEpochInteger to convert epoch
+				 * integers back to Iceberg binary.
 				 */
-				return psprintf("%s", col);
+				if (typeOid == DATEOID)
+					return psprintf("datediff('day', date '1970-01-01', %s::date)", col);
+				else if (typeOid == TIMESTAMPOID || typeOid == TIMESTAMPTZOID)
+					return psprintf("epoch_us(%s)", col);
+				else
+					return psprintf("%s", col);
 			}
 
 		case PARTITION_TRANSFORM_YEAR:
@@ -144,6 +150,85 @@ GetPartitionByExpressions(List *transforms)
 }
 
 
+/*
+ * HexDigitToInt converts a hex character ('0'-'9', 'A'-'F', 'a'-'f') to its
+ * integer value (0-15). Returns -1 for invalid characters.
+ */
+static int
+HexDigitToInt(char c)
+{
+	if (c >= '0' && c <= '9')
+		return c - '0';
+	if (c >= 'A' && c <= 'F')
+		return c - 'A' + 10;
+	if (c >= 'a' && c <= 'f')
+		return c - 'a' + 10;
+	return -1;
+}
+
+
+/*
+ * UrlDecodePartitionValue decodes percent-encoded characters in a Hive-style
+ * partition value (e.g. "1e%2B20" -> "1e+20").
+ *
+ * DuckDB percent-encodes special characters when writing partition directory
+ * names. We must decode them before parsing the value.
+ */
+static char *
+UrlDecodePartitionValue(const char *encoded)
+{
+	int			len = strlen(encoded);
+	char	   *decoded = palloc(len + 1);
+	int			j = 0;
+
+	for (int i = 0; i < len; i++)
+	{
+		if (encoded[i] == '%' && i + 2 < len)
+		{
+			int			hi = HexDigitToInt(encoded[i + 1]);
+			int			lo = HexDigitToInt(encoded[i + 2]);
+
+			if (hi >= 0 && lo >= 0)
+			{
+				decoded[j++] = (char) (hi * 16 + lo);
+				i += 2;
+				continue;
+			}
+		}
+		decoded[j++] = encoded[i];
+	}
+
+	decoded[j] = '\0';
+	return decoded;
+}
+
+
+/*
+ * NormalizeDuckDBTextToPGText converts a DuckDB text representation of a value
+ * to PostgreSQL's canonical text format by roundtripping through PG's type I/O.
+ *
+ * DuckDB may format values differently from PG (e.g. "1.0" vs "1" for numeric,
+ * "-0.0" vs "-0" for float8). This normalization ensures the text matches what
+ * DeserializePartitionValueFromPGText expects for its roundtrip assertion.
+ */
+static char *
+NormalizeDuckDBTextToPGText(const char *duckdbText, Oid resultTypeOid,
+							int32 resultTypeMod)
+{
+	Oid			typoinput;
+	Oid			typioparam;
+	Oid			typoutput;
+	bool		typIsVarlena;
+
+	getTypeInputInfo(resultTypeOid, &typoinput, &typioparam);
+	Datum		d = OidInputFunctionCall(typoinput, (char *) duckdbText,
+										 typioparam, resultTypeMod);
+
+	getTypeOutputInfo(resultTypeOid, &typoutput, &typIsVarlena);
+	return OidOutputFunctionCall(typoutput, d);
+}
+
+
 /*
  * ParsePartitionValuesFromPath extracts partition values from the Hive-style
  * directory path produced by DuckDB COPY TO with PARTITION_BY.
@@ -168,7 +253,7 @@ ParsePartitionValuesFromPath(char *filePath, List *transforms)
 		IcebergPartitionTransform *transform = list_nth(transforms, partIndex);
 
 		/* build the search key: "__part_N=" */
-		char	   *searchKey = psprintf(PARTITION_COLUMN_PREFIX "%d=", partIndex);
+		char	   *searchKey = psprintf("__part_%d=", partIndex);
 		int			searchKeyLen = strlen(searchKey);
 
 		/* find this key in the path */
@@ -190,6 +275,9 @@ ParsePartitionValuesFromPath(char *filePath, List *transforms)
 
 		char	   *valueText = pnstrdup(valueStart, valueLen);
 
+		/* URL-decode (DuckDB percent-encodes special chars in Hive paths) */
+		valueText = UrlDecodePartitionValue(valueText);
+
 		/* populate the partition field */
 		PartitionField *field = &partition->fields[partIndex];
 
@@ -203,12 +291,29 @@ ParsePartitionValuesFromPath(char *filePath, List *transforms)
 			field->value = NULL;
 			field->value_length = 0;
 		}
+		else if (transform->type == PARTITION_TRANSFORM_IDENTITY &&
+				 (transform->pgType.postgresTypeOid == DATEOID ||
+				  transform->pgType.postgresTypeOid == TIMESTAMPOID ||
+				  transform->pgType.postgresTypeOid == TIMESTAMPTZOID))
+		{
+			/*
+			 * Identity temporal types use epoch integers in the path (days
+			 * for date, microseconds for timestamp).
+			 */
+			field->value = DeserializePartitionValueFromEpochInteger(
+																	 transform, valueText, &field->value_length);
+		}
 		else
 		{
 			/*
-			 * Convert the text value to Iceberg binary format. The text
-			 * representation matches the transform's result PG type.
+			 * Normalize DuckDB text to PG canonical format (e.g. "1.0" -> "1"
+			 * for numeric) so the roundtrip assertion in
+			 * DeserializePartitionValueFromPGText passes.
 			 */
+			valueText = NormalizeDuckDBTextToPGText(valueText,
+													transform->resultPgType.postgresTypeOid,
+													transform->resultPgType.postgresTypeMod);
+
 			field->value = DeserializePartitionValueFromPGText(
 															   transform, valueText, &field->value_length);
 		}

pg_lake_table/src/fdw/partition_transform.c

@@ -38,6 +38,7 @@
 #include "pg_lake/iceberg/truncate_utils.h"
 #include "pg_lake/util/numeric.h"
 #include "pg_lake/util/rel_utils.h"
+#include "pg_lake/util/temporal_utils.h"
 #include "pg_lake/util/timetz.h"
 
 static PartitionField * ApplyPartitionTransformToTuple(IcebergPartitionTransform * transform,
@@ -1054,6 +1055,53 @@ DeserializePartitionValueFromPGText(IcebergPartitionTransform * transform,
 	return value;
 }
 
+/*
+ * DeserializePartitionValueFromEpochInteger converts an epoch integer text
+ * (from a DuckDB Hive-style partition path) to Iceberg binary format.
+ *
+ * For identity transforms on temporal types, the DuckDB expression produces
+ * epoch integers (days for date, microseconds for timestamp) instead of native
+ * text to avoid formatting issues with BC dates.
+ */
+void *
+DeserializePartitionValueFromEpochInteger(IcebergPartitionTransform * transform,
+										  const char *epochText, size_t *valueLength)
+{
+	if (epochText == NULL)
+	{
+		*valueLength = 0;
+		return NULL;
+	}
+
+	Oid			typeOid = transform->pgType.postgresTypeOid;
+	Datum		valueDatum;
+
+	if (typeOid == DATEOID)
+	{
+		int32		epochDays = pg_strtoint32(epochText);
+		DateADT		pgDate = AdjustDateFromUnixToPostgres(epochDays);
+
+		valueDatum = DateADTGetDatum(pgDate);
+	}
+	else if (typeOid == TIMESTAMPOID || typeOid == TIMESTAMPTZOID)
+	{
+		int64		epochMicros = pg_strtoint64(epochText);
+		Timestamp	pgTs = AdjustTimestampFromUnixToPostgres(epochMicros);
+
+		valueDatum = TimestampGetDatum(pgTs);
+	}
+	else
+	{
+		ereport(ERROR,
+				(errcode(ERRCODE_INTERNAL_ERROR),
+				 errmsg("DeserializePartitionValueFromEpochInteger called for non-temporal type %u",
+						typeOid)));
+	}
+
+	return DatumToPartitionValue(transform, valueDatum, false, valueLength);
+}
+
+
 /*
 * PartitionValueToDatum converts the partition value to a datum by deserializing
 * the value.