Use return_stats option to collect column statistics (#108)

Use duckdb output of COPY statements with return_stats option, to collect column statistics for parquet files.

Author: sfc-gh-agedemenli

pg_lake_copy/src/copy/copy.c

@@ -916,7 +916,7 @@ ProcessPgLakeCopyTo(CopyStmt *copyStmt, ParseState *pstate, Relation relation,
 	 */
 	ConvertCSVFileTo(tempCSVPath, tupleDesc, maximumLineLength,
 					 destinationPath, destinationFormat, destinationCompression,
-					 copyStmt->options, schema);
+					 copyStmt->options, schema, NIL);
 
 	if (IsCopyToStdout(copyStmt))
 	{

pg_lake_engine/include/pg_lake/data_file/data_file_stats.h

@@ -22,6 +22,36 @@
 #include "datatype/timestamp.h"
 
 #include "pg_lake/parquet/leaf_field.h"
+#include "pg_lake/pgduck/client.h"
+
+
+/*
+ * ColumnStatsMode describes the mode of column stats.
+ * - When truncate mode (default) is used, the column stats are truncated
+ *   to the given length.
+ * - When none mode is used, the column stats are not collected.
+ */
+typedef enum ColumnStatsMode
+{
+	COLUMN_STATS_MODE_TRUNCATE = 0,
+	COLUMN_STATS_MODE_NONE = 1,
+}			ColumnStatsMode;
+
+/*
+ * ColumnStatsConfig describes the configuration for column stats.
+ * - mode: the mode of column stats.
+ * - truncateLen: the length to truncate the column stats in truncate mode.
+ */
+typedef struct ColumnStatsConfig
+{
+	ColumnStatsMode mode;
+
+	/* used for truncate mode */
+	size_t		truncateLen;
+}			ColumnStatsConfig;
+
+
+
 
  /*
   * DataFileColumnStats stores column statistics for a data file.
@@ -43,6 +73,8 @@ typedef struct DataFileColumnStats
   */
 typedef struct DataFileStats
 {
+	char	   *dataFilePath;
+
 	/* number of bytes in the file */
 	int64		fileSize;
 
@@ -61,3 +93,26 @@ typedef struct DataFileStats
 	/* for a new data file with row IDs, the start of the range */
 	int64		rowIdStart;
 }			DataFileStats;
+
+typedef struct StatsCollector
+{
+	int64		totalRowCount;
+	List	   *dataFileStats;
+}			StatsCollector;
+
+extern PGDLLEXPORT DataFileStats * DeepCopyDataFileStats(const DataFileStats * stats);
+extern PGDLLEXPORT StatsCollector * GetDataFileStatsListFromPGResult(PGresult *result,
+																	 List *leafFields,
+																	 DataFileSchema * schema);
+extern PGDLLEXPORT StatsCollector * ExecuteCopyToCommandOnPGDuckConnection(char *copyCommand,
+																		   List *leafFields,
+																		   DataFileSchema * schema,
+																		   bool disablePreserveInsertionOrder,
+																		   char *destinationPath,
+																		   CopyDataFormat destinationFormat);
+extern PGDLLEXPORT bool ShouldSkipStatistics(LeafField * leafField);
+extern PGDLLEXPORT DataFileStats * CreateDataFileStatsForDataFile(char *dataFilePath,
+																  int64 rowCount, int64 deletedRowCount,
+																  List *leafFields);
+extern PGDLLEXPORT void ApplyColumnStatsModeForAllFileStats(Oid relationId, List *dataFileStats);
+extern PGDLLEXPORT List *GetRemoteParquetColumnStats(char *path, List *leafFields);

pg_lake_engine/include/pg_lake/parquet/field.h

@@ -34,6 +34,7 @@
 #pragma once
 
 #include "nodes/pg_list.h"
+#include "pg_lake/pgduck/type.h"
 
 /*
  * Reserved _row_id field ID used for Iceberg
@@ -154,3 +155,5 @@ typedef FieldStruct DataFileSchema;
 typedef FieldStructElement DataFileSchemaField;
 
 extern PGDLLEXPORT DataFileSchema * DeepCopyDataFileSchema(const DataFileSchema * schema);
+extern PGDLLEXPORT Field * DeepCopyField(const Field * field);
+extern PGDLLEXPORT bool PGTypeRequiresConversionToIcebergString(Field * field, PGType pgType);

pg_lake_engine/include/pg_lake/parquet/leaf_field.h

@@ -51,6 +51,8 @@ typedef struct LeafField
 
 extern PGDLLEXPORT int LeafFieldCompare(const ListCell *a, const ListCell *b);
 extern PGDLLEXPORT bool SchemaFieldsEquivalent(DataFileSchemaField * fieldA, DataFileSchemaField * fieldB);
+extern PGDLLEXPORT LeafField DeepCopyLeafField(const LeafField * leafField);
+extern PGDLLEXPORT LeafField * FindLeafField(List *leafFieldList, int fieldId);
 #if PG_VERSION_NUM < 170000
 extern PGDLLEXPORT int pg_cmp_s32(int32 a, int32 b);
 #endif

pg_lake_engine/include/pg_lake/pgduck/delete_data.h

@@ -22,11 +22,13 @@
 #include "pg_lake/copy/copy_format.h"
 #include "pg_lake/parquet/field.h"
 #include "pg_lake/pgduck/read_data.h"
+#include "pg_lake/data_file/data_file_stats.h"
 
-extern PGDLLEXPORT void PerformDeleteFromParquet(char *sourceDataFilePath,
-												 List *positionDeleteFiles,
-												 char *deletionFilePath,
-												 char *destinationPath,
-												 CopyDataCompression destinationCompression,
-												 DataFileSchema * schema,
-												 ReadDataStats * stats);
+extern PGDLLEXPORT StatsCollector * PerformDeleteFromParquet(char *sourceDataFilePath,
+															 List *positionDeleteFiles,
+															 char *deletionFilePath,
+															 char *destinationPath,
+															 CopyDataCompression destinationCompression,
+															 DataFileSchema * schema,
+															 ReadDataStats * stats,
+															 List *leafFields);

pg_lake_engine/include/pg_lake/pgduck/write_data.h

@@ -19,6 +19,7 @@
 
 #include "access/tupdesc.h"
 #include "pg_lake/copy/copy_format.h"
+#include "pg_lake/data_file/data_file_stats.h"
 #include "pg_lake/parquet/field.h"
 #include "nodes/pg_list.h"
 
@@ -35,20 +36,22 @@ typedef enum ParquetVersion
 /* pg_lake_table.default_parquet_version */
 extern PGDLLEXPORT int DefaultParquetVersion;
 
-extern PGDLLEXPORT void ConvertCSVFileTo(char *csvFilePath,
-										 TupleDesc tupleDesc,
-										 int maxLineSize,
-										 char *destinationPath,
-										 CopyDataFormat destinationFormat,
-										 CopyDataCompression destinationCompression,
-										 List *formatOptions,
-										 DataFileSchema * schema);
-extern PGDLLEXPORT int64 WriteQueryResultTo(char *query,
-											char *destinationPath,
-											CopyDataFormat destinationFormat,
-											CopyDataCompression destinationCompression,
-											List *formatOptions,
-											bool queryHasRowId,
-											DataFileSchema * schema,
-											TupleDesc queryTupleDesc);
+extern PGDLLEXPORT StatsCollector * ConvertCSVFileTo(char *csvFilePath,
+													 TupleDesc tupleDesc,
+													 int maxLineSize,
+													 char *destinationPath,
+													 CopyDataFormat destinationFormat,
+													 CopyDataCompression destinationCompression,
+													 List *formatOptions,
+													 DataFileSchema * schema,
+													 List *leafFields);
+extern PGDLLEXPORT StatsCollector * WriteQueryResultTo(char *query,
+													   char *destinationPath,
+													   CopyDataFormat destinationFormat,
+													   CopyDataCompression destinationCompression,
+													   List *formatOptions,
+													   bool queryHasRowId,
+													   DataFileSchema * schema,
+													   TupleDesc queryTupleDesc,
+													   List *leafFields);
 extern PGDLLEXPORT void AppendFields(StringInfo map, DataFileSchema * schema);

pg_lake_engine/pg_lake_engine--3.0--3.1.sql

@@ -33,3 +33,11 @@ CREATE FUNCTION __lake__internal__nsp__.from_hex(text)
  LANGUAGE C
  IMMUTABLE PARALLEL SAFE STRICT
 AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;
+
+-- Register map types, will be used for parsing DuckDB maps for COPY .. (return_stats)
+-- we prefer to create in the extension script to avoid concurrent attempts to create
+-- the same map, which may throw errors 
+WITH text_text_map_name AS
+ (SELECT map_type.create('TEXT','TEXT') AS name)
+SELECT map_type.create('TEXT', name) AS text_map_of_text
+ FROM text_text_map_name;