fix: add DuckDB dialect UDFs so flush partition filters run over SPI

pg_ducklake/include/pgducklake/pgducklake_metadata_manager.hpp

@@ -40,6 +40,8 @@ class PgDuckLakeMetadataManager : public duckdb::PostgresMetadataManager {
 	ReadAllInlinedDataForFlush(duckdb::DuckLakeSnapshot snapshot, const duckdb::string &inlined_table_name,
 	                           const duckdb::vector<duckdb::string> &columns_to_read) override;
 
+	duckdb::vector<duckdb::string> GetFlushPartitionSQLExpressions(const duckdb::DuckLakeTableEntry &table) override;
+
 	static bool IsInitialized();
 	// In-process SPI backend has no DuckDB metadata connection: the base AttachMetadata
 	// ATTACHes on it (null here) and MetadataExists probes ducklake_metadata in a way that

pg_ducklake/sql/pg_ducklake--1.0.0--1.1.0.sql

@@ -178,3 +178,134 @@ CREATE OPERATOR CLASS ducklake.unresolved_type_hash_ops
 DEFAULT FOR TYPE ducklake.unresolved_type USING hash AS
     OPERATOR 1 = (ducklake.unresolved_type, ducklake.unresolved_type),
     FUNCTION 1 ducklake.unresolved_type_hash(ducklake.unresolved_type);
+
+-- DuckDB-dialect partition transform functions (GitHub issue #223).
+--
+-- DuckLake generates flush_inlined_data()'s per-file "already deleted rows"
+-- bookkeeping query with the table's partition expression in DuckDB dialect:
+-- year(col)/month(col)/day(col)/hour(col) and, for bucket partitions,
+-- (murmur3_32(col) & 2147483647) % N. pg_ducklake executes that query in
+-- PostgreSQL over SPI (search_path forced to this schema), so these DuckDB
+-- function names must exist here with DuckDB-identical semantics. Each
+-- expression compares against the partition value the DuckDB writer baked
+-- into the data file, so any semantic drift mis-tracks deleted rows.
+
+-- Time transforms. DuckDB's year()/month()/day()/hour() return BIGINT; the
+-- timestamptz variants depend on the session TimeZone (like DuckDB's, which
+-- is synced from PostgreSQL when the embedded instance is created), hence
+-- STABLE rather than IMMUTABLE.
+CREATE FUNCTION ducklake.year(date) RETURNS bigint
+    SET search_path = pg_catalog, pg_temp
+    LANGUAGE sql IMMUTABLE STRICT
+    AS $$ SELECT EXTRACT(YEAR FROM $1)::bigint $$;
+CREATE FUNCTION ducklake.month(date) RETURNS bigint
+    SET search_path = pg_catalog, pg_temp
+    LANGUAGE sql IMMUTABLE STRICT
+    AS $$ SELECT EXTRACT(MONTH FROM $1)::bigint $$;
+CREATE FUNCTION ducklake.day(date) RETURNS bigint
+    SET search_path = pg_catalog, pg_temp
+    LANGUAGE sql IMMUTABLE STRICT
+    AS $$ SELECT EXTRACT(DAY FROM $1)::bigint $$;
+-- hour of a date is 0 (DuckDB implicitly casts date -> timestamp at midnight)
+CREATE FUNCTION ducklake.hour(date) RETURNS bigint
+    SET search_path = pg_catalog, pg_temp
+    LANGUAGE sql IMMUTABLE STRICT
+    AS $$ SELECT EXTRACT(HOUR FROM $1::timestamp)::bigint $$;
+
+CREATE FUNCTION ducklake.year(timestamp) RETURNS bigint
+    SET search_path = pg_catalog, pg_temp
+    LANGUAGE sql IMMUTABLE STRICT
+    AS $$ SELECT EXTRACT(YEAR FROM $1)::bigint $$;
+CREATE FUNCTION ducklake.month(timestamp) RETURNS bigint
+    SET search_path = pg_catalog, pg_temp
+    LANGUAGE sql IMMUTABLE STRICT
+    AS $$ SELECT EXTRACT(MONTH FROM $1)::bigint $$;
+CREATE FUNCTION ducklake.day(timestamp) RETURNS bigint
+    SET search_path = pg_catalog, pg_temp
+    LANGUAGE sql IMMUTABLE STRICT
+    AS $$ SELECT EXTRACT(DAY FROM $1)::bigint $$;
+CREATE FUNCTION ducklake.hour(timestamp) RETURNS bigint
+    SET search_path = pg_catalog, pg_temp
+    LANGUAGE sql IMMUTABLE STRICT
+    AS $$ SELECT EXTRACT(HOUR FROM $1)::bigint $$;
+
+-- DuckDB accepts only hour() on TIME (year/month/day fail to bind there)
+CREATE FUNCTION ducklake.hour(time) RETURNS bigint
+    SET search_path = pg_catalog, pg_temp
+    LANGUAGE sql IMMUTABLE STRICT
+    AS $$ SELECT EXTRACT(HOUR FROM $1)::bigint $$;
+
+CREATE FUNCTION ducklake.year(timestamptz) RETURNS bigint
+    SET search_path = pg_catalog, pg_temp
+    LANGUAGE sql STABLE STRICT
+    AS $$ SELECT EXTRACT(YEAR FROM $1)::bigint $$;
+CREATE FUNCTION ducklake.month(timestamptz) RETURNS bigint
+    SET search_path = pg_catalog, pg_temp
+    LANGUAGE sql STABLE STRICT
+    AS $$ SELECT EXTRACT(MONTH FROM $1)::bigint $$;
+CREATE FUNCTION ducklake.day(timestamptz) RETURNS bigint
+    SET search_path = pg_catalog, pg_temp
+    LANGUAGE sql STABLE STRICT
+    AS $$ SELECT EXTRACT(DAY FROM $1)::bigint $$;
+CREATE FUNCTION ducklake.hour(timestamptz) RETURNS bigint
+    SET search_path = pg_catalog, pg_temp
+    LANGUAGE sql STABLE STRICT
+    AS $$ SELECT EXTRACT(HOUR FROM $1)::bigint $$;
+
+-- Iceberg-compatible murmur3 x86 32-bit hash (seed 0), bit-exact with
+-- DuckLake's murmur3_32. One overload per exact input type: the generated
+-- SQL always casts the column to its DuckLake type, and exact matches keep
+-- PostgreSQL's function resolution from picking a wrong-encoding overload
+-- (e.g. an int must hash as a sign-extended 8-byte long, not as a double).
+CREATE FUNCTION ducklake.murmur3_32(boolean) RETURNS integer
+    SET search_path = pg_catalog, pg_temp
+    LANGUAGE C IMMUTABLE STRICT AS 'MODULE_PATHNAME', 'ducklake_murmur3_bool';
+CREATE FUNCTION ducklake.murmur3_32(smallint) RETURNS integer
+    SET search_path = pg_catalog, pg_temp
+    LANGUAGE C IMMUTABLE STRICT AS 'MODULE_PATHNAME', 'ducklake_murmur3_int2';
+CREATE FUNCTION ducklake.murmur3_32(integer) RETURNS integer
+    SET search_path = pg_catalog, pg_temp
+    LANGUAGE C IMMUTABLE STRICT AS 'MODULE_PATHNAME', 'ducklake_murmur3_int4';
+CREATE FUNCTION ducklake.murmur3_32(bigint) RETURNS integer
+    SET search_path = pg_catalog, pg_temp
+    LANGUAGE C IMMUTABLE STRICT AS 'MODULE_PATHNAME', 'ducklake_murmur3_int8';
+CREATE FUNCTION ducklake.murmur3_32(real) RETURNS integer
+    SET search_path = pg_catalog, pg_temp
+    LANGUAGE C IMMUTABLE STRICT AS 'MODULE_PATHNAME', 'ducklake_murmur3_float4';
+CREATE FUNCTION ducklake.murmur3_32(double precision) RETURNS integer
+    SET search_path = pg_catalog, pg_temp
+    LANGUAGE C IMMUTABLE STRICT AS 'MODULE_PATHNAME', 'ducklake_murmur3_float8';
+CREATE FUNCTION ducklake.murmur3_32(text) RETURNS integer
+    SET search_path = pg_catalog, pg_temp
+    LANGUAGE C IMMUTABLE STRICT AS 'MODULE_PATHNAME', 'ducklake_murmur3_text';
+-- raw bytes, for BLOB columns and VARCHAR columns' BYTEA storage
+CREATE FUNCTION ducklake.murmur3_32(bytea) RETURNS integer
+    SET search_path = pg_catalog, pg_temp
+    LANGUAGE C IMMUTABLE STRICT AS 'MODULE_PATHNAME', 'ducklake_murmur3_text';
+CREATE FUNCTION ducklake.murmur3_32(date) RETURNS integer
+    SET search_path = pg_catalog, pg_temp
+    LANGUAGE C IMMUTABLE STRICT AS 'MODULE_PATHNAME', 'ducklake_murmur3_date';
+CREATE FUNCTION ducklake.murmur3_32(time) RETURNS integer
+    SET search_path = pg_catalog, pg_temp
+    LANGUAGE C IMMUTABLE STRICT AS 'MODULE_PATHNAME', 'ducklake_murmur3_time';
+CREATE FUNCTION ducklake.murmur3_32(timestamp) RETURNS integer
+    SET search_path = pg_catalog, pg_temp
+    LANGUAGE C IMMUTABLE STRICT AS 'MODULE_PATHNAME', 'ducklake_murmur3_timestamp';
+CREATE FUNCTION ducklake.murmur3_32(timestamptz) RETURNS integer
+    SET search_path = pg_catalog, pg_temp
+    LANGUAGE C IMMUTABLE STRICT AS 'MODULE_PATHNAME', 'ducklake_murmur3_timestamp';
+-- DuckDB hashes a DECIMAL of width <= 18 as its unscaled value in an 8-byte
+-- long (DECIMAL(18,3) 12.5 hashes as 12500). Wider decimals hash their
+-- fixed-scale string form; the flush filter renders those as ::text instead
+-- of calling this overload. Without an exact numeric overload, PostgreSQL
+-- would resolve murmur3_32(numeric) to the double precision one (the
+-- preferred numeric type) and silently mis-hash every bucket.
+CREATE FUNCTION ducklake.murmur3_32(numeric) RETURNS integer
+    SET search_path = pg_catalog, pg_temp
+    LANGUAGE sql IMMUTABLE STRICT
+    AS $$ SELECT ducklake.murmur3_32(($1 * power(10::numeric, scale($1)))::bigint) $$;
+-- DuckDB hashes a UUID's lowercase hyphenated string form
+CREATE FUNCTION ducklake.murmur3_32(uuid) RETURNS integer
+    SET search_path = pg_catalog, pg_temp
+    LANGUAGE sql IMMUTABLE STRICT
+    AS $$ SELECT ducklake.murmur3_32($1::text) $$;

pg_ducklake/src/murmur3.cpp

@@ -0,0 +1,118 @@
+/*
+ * PG-side murmur3_32 for DuckLake's Iceberg-compatible bucket partition
+ * transform. flush_inlined_data()'s bookkeeping SQL embeds
+ * (murmur3_32(col) & 2147483647) % N and runs over SPI, so PostgreSQL needs
+ * the hash with byte-for-byte DuckLake semantics. The core hash is reused
+ * from the vendored header; each overload mirrors the input encoding of
+ * DuckLake's Murmur3ScalarFunction (ducklake_murmur3.cpp):
+ *   - bool/int16/int32/int64: sign-extended to int64, hashed as 8 bytes
+ *   - float/double: widened to double, -0.0 normalized, IEEE754 bits as int64
+ *   - text: raw UTF-8 bytes
+ *   - date: days since 1970-01-01 as int64 (PG stores days since 2000-01-01)
+ *   - time: microseconds since midnight (same encoding in PG and DuckDB)
+ *   - timestamp/timestamptz: microseconds since 1970-01-01 UTC (PG epoch 2000)
+ */
+
+#include <cstring>
+
+#include <common/ducklake_murmur3.hpp>
+
+extern "C" {
+#include "postgres.h"
+
+#if PG_VERSION_NUM >= 160000
+#include "varatt.h"
+#endif
+
+#include "fmgr.h"
+#include "utils/date.h"
+#include "utils/timestamp.h"
+}
+
+static constexpr int64_t PG_TO_UNIX_EPOCH_DAYS = POSTGRES_EPOCH_JDATE - UNIX_EPOCH_JDATE;
+static constexpr int64_t PG_TO_UNIX_EPOCH_USECS = PG_TO_UNIX_EPOCH_DAYS * USECS_PER_DAY;
+
+static int32_t
+MurmurInt64(int64_t val) {
+	return duckdb::DuckLakeMurmur3::HashValue<int64_t>(val);
+}
+
+static int32_t
+MurmurDouble(double val) {
+	if (val == 0.0) {
+		val = 0.0; // normalize -0.0, as DuckLake does
+	}
+	int64_t bits;
+	memcpy(&bits, &val, sizeof(bits));
+	return MurmurInt64(bits);
+}
+
+extern "C" {
+
+PG_FUNCTION_INFO_V1(ducklake_murmur3_bool);
+Datum
+ducklake_murmur3_bool(PG_FUNCTION_ARGS) {
+	PG_RETURN_INT32(MurmurInt64(PG_GETARG_BOOL(0) ? 1 : 0));
+}
+
+PG_FUNCTION_INFO_V1(ducklake_murmur3_int2);
+Datum
+ducklake_murmur3_int2(PG_FUNCTION_ARGS) {
+	PG_RETURN_INT32(MurmurInt64(PG_GETARG_INT16(0)));
+}
+
+PG_FUNCTION_INFO_V1(ducklake_murmur3_int4);
+Datum
+ducklake_murmur3_int4(PG_FUNCTION_ARGS) {
+	PG_RETURN_INT32(MurmurInt64(PG_GETARG_INT32(0)));
+}
+
+PG_FUNCTION_INFO_V1(ducklake_murmur3_int8);
+Datum
+ducklake_murmur3_int8(PG_FUNCTION_ARGS) {
+	PG_RETURN_INT32(MurmurInt64(PG_GETARG_INT64(0)));
+}
+
+PG_FUNCTION_INFO_V1(ducklake_murmur3_float4);
+Datum
+ducklake_murmur3_float4(PG_FUNCTION_ARGS) {
+	PG_RETURN_INT32(MurmurDouble((double)PG_GETARG_FLOAT4(0)));
+}
+
+PG_FUNCTION_INFO_V1(ducklake_murmur3_float8);
+Datum
+ducklake_murmur3_float8(PG_FUNCTION_ARGS) {
+	PG_RETURN_INT32(MurmurDouble(PG_GETARG_FLOAT8(0)));
+}
+
+/* Backs both the text and bytea overloads: DuckLake hashes a VARCHAR's UTF-8
+ * bytes and a BLOB's raw bytes, the same raw-varlena-bytes rule. */
+PG_FUNCTION_INFO_V1(ducklake_murmur3_text);
+Datum
+ducklake_murmur3_text(PG_FUNCTION_ARGS) {
+	text *val = PG_GETARG_TEXT_PP(0);
+	auto data = reinterpret_cast<const uint8_t *>(VARDATA_ANY(val));
+	PG_RETURN_INT32(duckdb::DuckLakeMurmur3::Hash(data, VARSIZE_ANY_EXHDR(val)));
+}
+
+PG_FUNCTION_INFO_V1(ducklake_murmur3_date);
+Datum
+ducklake_murmur3_date(PG_FUNCTION_ARGS) {
+	PG_RETURN_INT32(MurmurInt64((int64_t)PG_GETARG_DATEADT(0) + PG_TO_UNIX_EPOCH_DAYS));
+}
+
+PG_FUNCTION_INFO_V1(ducklake_murmur3_time);
+Datum
+ducklake_murmur3_time(PG_FUNCTION_ARGS) {
+	PG_RETURN_INT32(MurmurInt64(PG_GETARG_TIMEADT(0)));
+}
+
+/* Timestamp and TimestampTz share the representation (microseconds since
+ * 2000-01-01 UTC), so this one entry point backs both SQL overloads. */
+PG_FUNCTION_INFO_V1(ducklake_murmur3_timestamp);
+Datum
+ducklake_murmur3_timestamp(PG_FUNCTION_ARGS) {
+	PG_RETURN_INT32(MurmurInt64(PG_GETARG_TIMESTAMP(0) + PG_TO_UNIX_EPOCH_USECS));
+}
+
+} // extern "C"

pg_ducklake/src/pgducklake_metadata_manager.cpp

@@ -22,6 +22,9 @@
 #include <duckdb/common/types/value.hpp>
 #include <duckdb/main/client_context.hpp>
 #include <duckdb/main/materialized_query_result.hpp>
+#include <duckdb/parser/keyword_helper.hpp>
+#include <storage/ducklake_partition_data.hpp>
+#include <storage/ducklake_table_entry.hpp>
 
 extern "C" {
 #include "postgres.h"
@@ -96,6 +99,9 @@ SPIExecuteInSubtransaction(const duckdb::string &query, bool &had_error, duckdb:
 	/* Suppress NOTICEs: DuckLake re-runs CREATE TABLE IF NOT EXISTS, whose NOTICE would leak to the client. */
 	int save_nestlevel = NewGUCNestLevel();
 	::SetConfigOption("client_min_messages", "warning", PGC_USERSET, PGC_S_SESSION);
+	/* DuckLake-generated SQL calls DuckDB-dialect functions (month(), murmur3_32(), ...) unqualified;
+	 * resolve them to the ducklake-schema UDFs deterministically, independent of the caller's search_path. */
+	::SetConfigOption("search_path", "ducklake", PGC_USERSET, PGC_S_SESSION);
 
 	SetAllowSubtransaction(true);
 	BeginInternalSubTransaction(NULL);
@@ -351,6 +357,53 @@ ORDER BY row_id, begin_snapshot;)",
 	return transaction.ExecuteRaw(query);
 }
 
+/*
+ * The flush's deleted-rows filter runs over SPI against the inlined heap table, so each partition
+ * column must be rendered as PG SQL that recovers the DuckLake-typed value from its storage type:
+ *  - VARCHAR is stored as BYTEA; the upstream CAST(col AS VARCHAR) would yield PG's hex form
+ *    ('\x6170706c65'), silently mis-hashing every bucket/identity comparison. convert_from()
+ *    restores the original string.
+ *  - BLOB is stored as BYTEA; pass it through raw (murmur3_32(bytea) hashes the bytes directly,
+ *    matching DuckDB, which hashes a BLOB's raw bytes).
+ *  - Types stored as VARCHAR (date/timestamp family) keep the upstream CAST to their DuckDB type
+ *    name, which PG parses from the stored text form.
+ *  - Native types cast via GetColumnTypeInternal so the type name is PG-parseable (e.g. DuckDB
+ *    DOUBLE -> DOUBLE PRECISION).
+ * The transform wrapper (month(...), (murmur3_32(...) & ...) % N) then resolves to the ducklake
+ * schema UDFs via the search_path forced in SPIExecuteInSubtransaction.
+ */
+duckdb::vector<duckdb::string>
+PgDuckLakeMetadataManager::GetFlushPartitionSQLExpressions(const duckdb::DuckLakeTableEntry &table) {
+	duckdb::vector<duckdb::string> result;
+	auto partition_data = table.GetPartitionData();
+	if (!partition_data) {
+		return result;
+	}
+	for (auto &field : partition_data->fields) {
+		auto &col = table.GetColumnByFieldId(field.field_id);
+		auto col_name = duckdb::KeywordHelper::WriteOptionallyQuoted(col.GetName());
+		auto type = col.GetType();
+		duckdb::string rendered;
+		if (type.id() == duckdb::LogicalTypeId::VARCHAR) {
+			rendered = "convert_from(" + col_name + ", 'UTF8')";
+		} else if (type.id() == duckdb::LogicalTypeId::BLOB) {
+			rendered = col_name;
+		} else if (GetColumnTypeInternal(type) == "VARCHAR") {
+			rendered = "CAST(" + col_name + " AS " + type.ToString() + ")";
+		} else {
+			rendered = "CAST(" + col_name + " AS " + GetColumnTypeInternal(type) + ")";
+		}
+		/* DuckDB hashes DECIMALs wider than 18 digits (INT128 storage) by their fixed-scale
+		 * string form; ducklake.murmur3_32(numeric) implements the narrow unscaled-long rule. */
+		if (field.transform.type == duckdb::DuckLakeTransformType::BUCKET &&
+		    type.id() == duckdb::LogicalTypeId::DECIMAL && duckdb::DecimalType::GetWidth(type) > 18) {
+			rendered = "CAST(" + rendered + " AS TEXT)";
+		}
+		result.push_back(duckdb::DuckLakePartitionUtils::GetPartitionSQLExpression(field.transform, rendered));
+	}
+	return result;
+}
+
 duckdb::unique_ptr<duckdb::QueryResult>
 PgDuckLakeMetadataManager::Query(duckdb::DuckLakeSnapshot snapshot, duckdb::string query) {
 	DuckLakeMetadataManager::FillSnapshotArgs(query, snapshot);