feat: maintain inlined-data column stats in the direct-insert writer

pg_ducklake/include/pgducklake/inline_col_stats.hpp

@@ -0,0 +1,54 @@
+#pragma once
+
+#include "pgddb/pg/declarations.hpp"
+#include "pgducklake/pgducklake_metadata_manager.hpp"
+
+#include <memory>
+#include <vector>
+
+namespace pgducklake {
+
+/* Per-column min/max + null accumulator for a single inline-insert batch (direct insert or COPY
+ * FROM).
+ * Cells are folded as native PG Datums and compared type-correctly in PG's own value space
+ * (numeric/temporal/boolean via the type's btree comparison proc; VARCHAR/UUID lexicographically
+ * over the canonical output text) -- mirroring DuckLakeColumnStats::MergeStats without a
+ * per-cell text->Value->cast->ToString round-trip. Only the two surviving bounds per column are
+ * canonicalized (via duckdb::Value::ToString) at Finalize(), so the persisted encoding round-trips
+ * through the read path exactly as native DuckLake's inline write would produce it.
+ * CreateSnapshotForDirectInsert widens those bounds into ducklake_table_column_stats.
+ * The DuckDB-typed and PG-typed state lives behind a pimpl so this header stays free of DuckDB and
+ * postgres.h includes. */
+class InlineColStats {
+public:
+	/* Fetches column_id + DuckLake type for the first num_cols user columns (ordered by
+	 * column_order -- both the direct-insert and COPY paths write a prefix of the table columns
+	 * in that order). MUST be called inside an already-open SPI connection. On any metadata
+	 * shortfall the accumulator is inactive and no stats are maintained (existing bounds stay
+	 * intact -- never widened into a lie). */
+	InlineColStats(uint64_t table_id, int num_cols);
+	~InlineColStats();
+
+	InlineColStats(const InlineColStats &) = delete;
+	InlineColStats &operator=(const InlineColStats &) = delete;
+
+	bool ColumnEligible(int col) const;
+	/* Bind an eligible column's PG source type so cells can be compared natively. Call once per
+	 * column before the row loop; a no-op for ineligible columns. If the source type has no usable
+	 * comparison the column is silently demoted to ineligible (no stats, never a lie). Needs no SPI
+	 * connection. */
+	void SetupColumn(int col, Oid pg_source_type);
+	/* Fold one non-null cell, given as a PG Datum of the column's bound source type. */
+	void ObserveDatum(int col, Datum value);
+	/* Not gated on ColumnEligible: null-ness needs no comparison proc or canonical encoding, so it
+	 * costs nothing to match the normal path here. The min/max exclusions are not a precedent --
+	 * they exist only because a canonical bound for those types is not cheap to produce. */
+	void ObserveNull(int col);
+	void Finalize(std::vector<DirectInsertColumnStat> &out);
+
+private:
+	struct Impl;
+	std::unique_ptr<Impl> impl;
+};
+
+} // namespace pgducklake

pg_ducklake/include/pgducklake/pgducklake_metadata_manager.hpp

@@ -2,6 +2,9 @@
 
 #include "pgddb/pg/declarations.hpp"
 
+#include <string>
+#include <vector>
+
 #include <common/ducklake_encryption.hpp>
 #include <common/ducklake_options.hpp>
 #include <common/ducklake_snapshot.hpp>
@@ -82,6 +85,23 @@ bool GetTableInliningInfo(Oid table_oid, uint64_t *table_id_out, uint64_t *schem
 
 uint64_t GetNextRowIdForTable(uint64_t table_id, uint64_t schema_version);
 uint64_t GetNextSnapshotId();
-void CreateSnapshotForDirectInsert(uint64_t snapshot_id, uint64_t table_id, int64_t rows_inserted);
+
+/* Per-column min/max + null contribution of a single direct-insert batch. min_value/max_value hold
+ * the DuckLake-canonical text encoding (duckdb::Value::ToString), so they round-trip through the read
+ * path's Value(text)->cast(column_type). column_type is the DuckLake type string
+ * (duckdb::DuckLakeTypes::FromString input) used to compare type-correctly when widening. */
+struct DirectInsertColumnStat {
+	uint64_t column_id = 0;
+	std::string column_type;
+	bool has_min = false;
+	bool has_max = false;
+	std::string min_value;
+	std::string max_value;
+	/* May be set with no bound at all: an all-NULL batch still has to flip contains_null. */
+	uint64_t null_count = 0;
+};
+
+void CreateSnapshotForDirectInsert(uint64_t snapshot_id, uint64_t table_id, int64_t rows_inserted,
+                                   const std::vector<DirectInsertColumnStat> &column_stats);
 
 } // namespace pgducklake

pg_ducklake/src/copy_from.cpp

@@ -4,8 +4,11 @@
 
 #include "pgducklake/catalog_sync.hpp"
 #include "pgducklake/copy_from.hpp"
+#include "pgducklake/inline_col_stats.hpp"
 #include "pgducklake/pgducklake_metadata_manager.hpp"
 
+#include <vector>
+
 extern "C" {
 #include "postgres.h"
 
@@ -16,6 +19,7 @@ extern "C" {
 #include "catalog/namespace.h"
 #include "commands/copy.h"
 #include "executor/executor.h"
+#include "executor/spi.h"
 #include "fmgr.h"
 #include "parser/parse_node.h"
 #include "utils/builtins.h"
@@ -113,6 +117,26 @@ DucklakeCopyFromStdin(CopyStmt *stmt, const char *query_string) {
 
 	ColumnConvInfo *conv = BuildColumnConvInfo(user_tupdesc, inlined_tupdesc);
 
+	/* Per-column min/max accumulator, widened into ducklake_table_column_stats after the batch so a
+	 * later reader never mis-prunes the inlined rows. Constructed under a scoped SPI connection (it
+	 * reads column metadata); its state is SPI-independent afterwards. Eligible columns stored
+	 * natively need their own output function to canonicalize each cell. */
+	if (SPI_connect() < 0) {
+		table_close(inlined_rel, RowExclusiveLock);
+		table_close(user_rel, RowExclusiveLock);
+		ereport(ERROR, (errcode(ERRCODE_INTERNAL_ERROR), errmsg("SPI_connect failed")));
+	}
+	InlineColStats col_stats(table_id, natts);
+	SPI_finish();
+
+	/* Bind the accumulator to each column's PG source type: cells fold as native Datums, compared in
+	 * PG value space -- no per-cell text/Value/cast round-trip. */
+	for (int i = 0; i < natts; i++) {
+		if (col_stats.ColumnEligible(i)) {
+			col_stats.SetupColumn(i, TupleDescAttr(user_tupdesc, i)->atttypid);
+		}
+	}
+
 	uint64_t begin_snapshot = GetNextSnapshotId();
 	uint64_t next_row_id = GetNextRowIdForTable(table_id, schema_version);
 
@@ -155,14 +179,17 @@ DucklakeCopyFromStdin(CopyStmt *stmt, const char *query_string) {
 		for (int i = 0; i < natts; i++) {
 			int dst = i + INLINED_SYSTEM_COLS;
 			if (copy_nulls[i]) {
+				col_stats.ObserveNull(i);
 				slot_values[dst] = (Datum)0;
 				slot_isnull[dst] = true;
 			} else if (conv[i].needs_text_conv) {
 				char *str = OutputFunctionCall(&conv[i].typoutput_finfo, copy_values[i]);
+				col_stats.ObserveDatum(i, copy_values[i]);
 				slot_values[dst] = CStringGetTextDatum(str);
 				slot_isnull[dst] = false;
 				pfree(str);
 			} else {
+				col_stats.ObserveDatum(i, copy_values[i]);
 				slot_values[dst] = copy_values[i];
 				slot_isnull[dst] = false;
 			}
@@ -208,8 +235,10 @@ DucklakeCopyFromStdin(CopyStmt *stmt, const char *query_string) {
 	table_close(user_rel, RowExclusiveLock);
 
 	if (rows_inserted > 0) {
+		std::vector<DirectInsertColumnStat> col_stats_out;
+		col_stats.Finalize(col_stats_out);
 		SkipSnapshotSyncGuard sync_guard;
-		CreateSnapshotForDirectInsert(begin_snapshot, table_id, rows_inserted);
+		CreateSnapshotForDirectInsert(begin_snapshot, table_id, rows_inserted, col_stats_out);
 		CommandCounterIncrement();
 	}
 

pg_ducklake/src/direct_insert.cpp

@@ -5,10 +5,13 @@
 #include "pgducklake/direct_insert.hpp"
 #include "pgducklake/duckdb_manager.hpp"
 #include "pgducklake/guc.hpp"
+#include "pgducklake/inline_col_stats.hpp"
 #include "pgducklake/pgducklake_metadata_manager.hpp"
 
 #include <cstring>
+#include <string>
 #include <unordered_map>
+#include <vector>
 
 #include <duckdb.hpp>
 
@@ -194,8 +197,10 @@ static bool IsUnnestOfParam(Node *node, int *param_id_out, Oid *param_type_out);
 static bool ValidateArrayLengths(ParamListInfo bound_params, List *param_ids, int *expected_row_count_out);
 static PlannedStmt *CreateDirectInsertPlan(Query *parse, DirectInsertContext *context);
 static PlannedStmt *CreateValuesInsertPlan(Query *parse, ValuesInsertContext *context);
-static void DirectInsertIntoInlinedTable(DirectInsertScanState *state);
-static void DirectInsertValuesIntoInlinedTable(DirectInsertScanState *state);
+static void DirectInsertIntoInlinedTable(DirectInsertScanState *state,
+                                         std::vector<DirectInsertColumnStat> &col_stats_out);
+static void DirectInsertValuesIntoInlinedTable(DirectInsertScanState *state,
+                                               std::vector<DirectInsertColumnStat> &col_stats_out);
 
 /*
  * Map a DuckDB type string to the PG OID used in the inlined data table;
@@ -1217,14 +1222,16 @@ DirectInsert_ExecCustomScan(CustomScanState *node) {
 		state->next_row_id = pgducklake::GetNextRowIdForTable(state->table_id, state->schema_version);
 		state->rows_inserted = 0;
 
+		std::vector<DirectInsertColumnStat> col_stats;
 		if (state->mode == DIRECT_INSERT_UNNEST) {
-			DirectInsertIntoInlinedTable(state);
+			DirectInsertIntoInlinedTable(state, col_stats);
 		} else {
-			DirectInsertValuesIntoInlinedTable(state);
+			DirectInsertValuesIntoInlinedTable(state, col_stats);
 		}
 
 		pgducklake::SkipSnapshotSyncGuard sync_guard;
-		pgducklake::CreateSnapshotForDirectInsert(state->begin_snapshot, state->table_id, state->rows_inserted);
+		pgducklake::CreateSnapshotForDirectInsert(state->begin_snapshot, state->table_id, state->rows_inserted,
+		                                          col_stats);
 	}
 	PG_CATCH();
 	{
@@ -1277,7 +1284,7 @@ DirectInsert_ExplainCustomScan(CustomScanState *node, List *ancestors, ExplainSt
 }
 
 static void
-DirectInsertIntoInlinedTable(DirectInsertScanState *state) {
+DirectInsertIntoInlinedTable(DirectInsertScanState *state, std::vector<DirectInsertColumnStat> &col_stats_out) {
 	int ret;
 
 	if ((ret = SPI_connect()) < 0) {
@@ -1381,6 +1388,16 @@ DirectInsertIntoInlinedTable(DirectInsertScanState *state) {
 		}
 	}
 
+	/* Per-column min/max accumulator: widened into the catalog after the batch is written. Cells are
+	 * folded as native PG Datums (element_types[i]) and compared in PG value space -- no per-cell
+	 * text/Value/cast round-trip. */
+	InlineColStats col_stats(state->table_id, num_params);
+	for (int i = 0; i < num_params; i++) {
+		if (col_stats.ColumnEligible(i)) {
+			col_stats.SetupColumn(i, element_types[i]);
+		}
+	}
+
 	uint64_t current_row_id = state->next_row_id;
 
 	for (int row = 0; row < arr_length; row++) {
@@ -1389,17 +1406,20 @@ DirectInsertIntoInlinedTable(DirectInsertScanState *state) {
 
 		for (int i = 0; i < num_params; i++) {
 			if (elem_nulls[i][row]) {
+				col_stats.ObserveNull(i);
 				values[i + 2] = (Datum)0;
 				nulls[i + 2] = 'n';
 			} else if (needs_text_conv[i]) {
 				// Scalar type (DATE, TIMESTAMP, UBIGINT, etc.) -> VARCHAR:
 				// use PG output function to produce a DuckDB-parseable text string.
 				char *str = OidOutputFunctionCall(typoutput[i], elem_values[i][row]);
+				col_stats.ObserveDatum(i, elem_values[i][row]);
 				values[i + 2] = CStringGetTextDatum(str);
 				nulls[i + 2] = ' ';
 				pfree(str);
 			} else {
 				// Types match (native), BYTEA zero-copy, or unexpected -- pass as-is.
+				col_stats.ObserveDatum(i, elem_values[i][row]);
 				values[i + 2] = elem_values[i][row];
 				nulls[i + 2] = ' ';
 			}
@@ -1413,6 +1433,8 @@ DirectInsertIntoInlinedTable(DirectInsertScanState *state) {
 		state->rows_inserted++;
 	}
 
+	col_stats.Finalize(col_stats_out);
+
 	SPI_finish();
 
 	ereport(DEBUG1, (errmsg("DuckLake direct insert: successfully inserted %lld rows into %s",
@@ -1431,11 +1453,21 @@ struct ValuesColumnConvInfo {
 };
 
 static void
-DirectInsertValuesIntoInlinedTable(DirectInsertScanState *state) {
+DirectInsertValuesIntoInlinedTable(DirectInsertScanState *state, std::vector<DirectInsertColumnStat> &col_stats_out) {
 	int num_rows = state->values_num_rows;
 	int num_cols = state->values_num_cols;
 	ExprContext *econtext = state->css.ss.ps.ps_ExprContext;
 
+	/* Per-column min/max accumulator (widened into the catalog after the batch). The VALUES path
+	 * writes via the table AM (no SPI), so build the accumulator -- which reads column metadata via
+	 * SPI in its constructor -- under a scoped SPI connection; its state is SPI-independent after. */
+	int sret;
+	if ((sret = SPI_connect()) < 0) {
+		ereport(ERROR, (errcode(ERRCODE_INTERNAL_ERROR), errmsg("SPI_connect failed: %d", sret)));
+	}
+	InlineColStats col_stats(state->table_id, num_cols);
+	SPI_finish();
+
 	char relname[NAMEDATALEN];
 	snprintf(relname, sizeof(relname), "ducklake_inlined_data_%llu_%llu", (unsigned long long)state->table_id,
 	         (unsigned long long)state->schema_version);
@@ -1474,6 +1506,12 @@ DirectInsertValuesIntoInlinedTable(DirectInsertScanState *state) {
 		} else {
 			conv[i].needs_text_conv = false;
 		}
+
+		/* Bind the accumulator to this column's PG source type: cells fold as native Datums,
+		 * compared in PG value space (no per-cell text/Value/cast round-trip). */
+		if (col_stats.ColumnEligible(i)) {
+			col_stats.SetupColumn(i, src_type);
+		}
 	}
 
 	/* Ensure DateStyle is ISO for temporal -> VARCHAR text conversion.
@@ -1529,14 +1567,17 @@ DirectInsertValuesIntoInlinedTable(DirectInsertScanState *state) {
 			Datum d = ExecEvalExprSwitchContext(state->values_estates[flat], econtext, &isnull);
 
 			if (isnull) {
+				col_stats.ObserveNull(col);
 				sv[dst] = (Datum)0;
 				sn[dst] = true;
 			} else if (conv[col].needs_text_conv) {
 				char *str = OutputFunctionCall(&conv[col].typoutput_finfo, d);
+				col_stats.ObserveDatum(col, d);
 				sv[dst] = CStringGetTextDatum(str);
 				sn[dst] = false;
 				pfree(str);
 			} else {
+				col_stats.ObserveDatum(col, d);
 				sv[dst] = d;
 				sn[dst] = false;
 			}
@@ -1577,6 +1618,8 @@ DirectInsertValuesIntoInlinedTable(DirectInsertScanState *state) {
 
 	state->rows_inserted = num_rows;
 
+	col_stats.Finalize(col_stats_out);
+
 	ereport(DEBUG1, (errmsg("DuckLake direct insert (VALUES): inserted %d rows into %s", num_rows, relname)));
 }