datalake_fdw: read and write Parquet through Arrow

.github/workflows/build-cloudberry.yml

@@ -1463,6 +1463,34 @@ jobs:
             exit 0
           fi
 
+          # datalake_fdw needs the Arrow and Parquet C++ libraries, which the
+          # build image does not carry -- nothing in the RPM uses them, so they
+          # would be weight every other job paid for.
+          if [[ "${PGXS_EXTENSION}" == "contrib/datalake_fdw" ]]; then
+            . /etc/os-release
+            if [[ "${VERSION_ID%%.*}" == "8" ]]; then
+              # EPEL 8 has them, but its libarrow-devel needs a utf8proc-devel
+              # that modular filtering keeps out of PowerTools, so it cannot be
+              # installed.  The Arrow project's own repository can.  Pinned to
+              # the version EPEL 10 carries, both because that is one version
+              # fewer to have working and because the newest wants C++20, which
+              # Rocky 8's gcc 8 does not have.
+              # EPEL as well, and not only for Arrow itself: arrow-devel needs
+              # re2-devel and parquet-devel needs thrift-devel, and on EL8 both
+              # of those live in EPEL.
+              dnf install -y \
+                https://apache.jfrog.io/artifactory/arrow/almalinux/8/apache-arrow-release-latest.rpm
+              dnf install -y --enablerepo=epel --enablerepo=powertools \
+                arrow-devel-17.0.0-1.el8 parquet-devel-17.0.0-1.el8
+            else
+              # From EPEL, which the image has enrolled but left disabled,
+              # exactly as it does for its own EPEL packages; CRB carries what
+              # they depend on.
+              dnf install -y --enablerepo=epel --enablerepo=crb \
+                libarrow-devel parquet-libs-devel
+            fi
+          fi
+
           # The RPM installs as root; the build runs as gpadmin, as everywhere
           # else in this job.  -H follows the command-line symlink.
           chown -RH gpadmin:gpadmin "${BUILD_DESTINATION}/"

contrib/datalake_fdw/Makefile

@@ -18,8 +18,11 @@
 # contrib/datalake_fdw/Makefile
 
 MODULE_big = datalake_fdw
-EXTENSION = datalake_fdw
-DATA = datalake_fdw--1.0.sql
+
+# A second extension, so that installing datalake_fdw does not put the test
+# entry points in a production database.  One library, so they reach internals.
+EXTENSION = datalake_fdw datalake_fdw_test
+DATA = datalake_fdw--1.0.sql datalake_fdw_test--1.0.sql
 
 OBJS = \
 	src/am_iceberg/pg_iceberg_am_handler.o \
@@ -36,20 +39,54 @@ OBJS = \
 	src/meta/meta_engine_init.o \
 	src/meta/engine_stub/stub_engine.o \
 	src/format/format_registry.o \
+	src/format/arrow_support.o \
+	src/format/arrow_builder.o \
+	src/format/arrow_decode.o \
+	src/format/parquet/parquet_format.o \
+	src/format/parquet/parquet_read.o \
+	src/format/parquet/parquet_write.o \
 	src/common/dl_err.o \
+	src/common/dl_resource.o \
 	src/common/dl_option_util.o \
 	src/common/parser_option.o \
 	src/common/file_system_wrapper.o \
 	src/common/s3_file_system.o \
-	src/common/backend_registry.o
+	src/common/backend_registry.o \
+	src/test/datalake_fdw_test.o
+
+# libparquet is written in terms of Arrow's types, so linking one links both.
+PKG_CONFIG ?= pkg-config
+ARROW_MODULES = arrow parquet
+HAVE_ARROW := $(shell $(PKG_CONFIG) --exists $(ARROW_MODULES) 2>/dev/null && echo yes)
+# Without the filter, whichever C++ standard Arrow's .pc file names wins: these
+# land in CPPFLAGS, which pgxs.mk puts after CXXFLAGS on the command line.  The
+# Arrow project's own packages say -std=c++11, and their headers then fail to
+# compile against themselves.
+ARROW_CPPFLAGS := $(filter-out -std=%,\
+	$(shell $(PKG_CONFIG) --cflags $(ARROW_MODULES) 2>/dev/null))
+ARROW_LIBS := $(shell $(PKG_CONFIG) --libs $(ARROW_MODULES) 2>/dev/null)
+
+# This has to refuse at parse time.  A recipe hung off `all` would run after
+# pgxs.mk's own all-lib, so the compiler would fail on a missing arrow/api.h
+# first and this would never be reached.  Not for the clean targets, because
+# contrib/Makefile recurses here for those even when the module is not
+# configured in.
+ifneq ($(HAVE_ARROW),yes)
+ifeq ($(filter clean distclean maintainer-clean,$(MAKECMDGOALS)),)
+$(error datalake_fdw needs the Apache Arrow and Parquet C++ libraries, and \
+pkg-config found neither "arrow" nor "parquet".  They are libarrow-devel and \
+parquet-libs-devel on Rocky and RHEL, libarrow-dev and libparquet-dev on \
+Debian and Ubuntu)
+endif
+endif
 
 # Use the documented PGXS knobs: pgxs.mk appends these AFTER the flags configure
 # chose, so optimization/warning settings survive.  A pre-include
 # "override CFLAGS +=" would give CFLAGS override origin and silently discard
 # Makefile.global's own "CFLAGS = @CFLAGS@" assignment.
 PG_CFLAGS = -fvisibility=hidden
 PG_CXXFLAGS = -fvisibility=hidden -fvisibility-inlines-hidden -std=c++17
-PG_CPPFLAGS = -I$(srcdir)/src
+PG_CPPFLAGS = -I$(srcdir)/src $(ARROW_CPPFLAGS)
 
 # The regression cases live with the rest of the test material rather than in a
 # second place of their own; pg_regress is pointed at them.  REGRESS_OPTS is
@@ -65,6 +102,10 @@ REGRESS = iceberg_am_ddl iceberg_am_reject iceberg_am_acl
 REGRESS_OPTS = --temp-config=$(srcdir)/datalake_fdw.conf \
 	--inputdir=$(srcdir)/test/automation/sqlrepo/smoke/iceberg_am
 
+# A second category, and pg_regress takes one --inputdir, so it is a second run.
+FORMAT_PARQUET_REGRESS = parquet_roundtrip
+FORMAT_PARQUET_INPUTDIR = $(srcdir)/test/automation/sqlrepo/smoke/format_parquet
+
 EXTRA_CLEAN = exports_darwin.list exports.map
 
 # Keep the aggregate target as make's default goal.
@@ -87,10 +128,10 @@ endif
 
 # Shared libraries are linked with $(CC) (see src/Makefile.shlib COMPILER), so a
 # module containing C++ translation units must pull in the C++ runtime itself.
-SHLIB_LINK += -lstdc++
+SHLIB_LINK += -lstdc++ $(ARROW_LIBS)
 
-# Arrow and other C++ dependencies land in this module later; the export list is
-# the single place that decides what stays visible, so the mechanism ships now.
+# The export list is the single place that decides what stays visible -- which
+# now also means none of Arrow's symbols become symbols this module offers.
 ifeq ($(PORTNAME), darwin)
 EXPORT_LIST = exports_darwin.list
 SHLIB_LINK += -Wl,-exported_symbols_list,exports_darwin.list
@@ -107,3 +148,30 @@ endif
 
 all: $(EXPORT_LIST)
 $(shlib): $(EXPORT_LIST)
+
+# Hung off check and installcheck so that both get both categories.  REGRESS_OPTS
+# has to come along: pgxs.mk is where --dbname=$(CONTRIB_TESTDB) is added to it,
+# and without that pg_regress falls back to "regression" -- which it DROPs and
+# recreates, taking the core suite's database with it.  The second --inputdir
+# wins over the one in REGRESS_OPTS.  submake and REGRESS_PREP are the same
+# prerequisites pgxs.mk gives its own targets, so that a parallel make cannot
+# start pg_regress before it has been built.
+installcheck: installcheck-format-parquet
+
+installcheck-format-parquet: submake $(REGRESS_PREP)
+	$(pg_regress_installcheck) $(REGRESS_OPTS) \
+		--inputdir=$(FORMAT_PARQUET_INPUTDIR) $(FORMAT_PARQUET_REGRESS)
+
+.PHONY: installcheck-format-parquet
+
+# "make check" is in-tree only -- under PGXS pgxs.mk refuses the target -- and
+# it is the only run that supplies the temp-config that preloads this module.
+ifndef USE_PGXS
+check: check-format-parquet
+
+check-format-parquet: submake $(REGRESS_PREP)
+	$(pg_regress_check) $(REGRESS_OPTS) \
+		--inputdir=$(FORMAT_PARQUET_INPUTDIR) $(FORMAT_PARQUET_REGRESS)
+
+.PHONY: check-format-parquet
+endif

contrib/datalake_fdw/datalake_fdw_test--1.0.sql

@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ * contrib/datalake_fdw/datalake_fdw_test--1.0.sql
+ */
+
+-- complain if script is sourced in psql, rather than via CREATE EXTENSION
+\echo Use "CREATE EXTENSION datalake_fdw_test" to load this file. \quit
+
+/*
+ * Both functions name a path on the server's file system and run as the
+ * operating system user the server does, so they are as privileged as
+ * pg_read_server_files and are granted the same way: to nobody, until someone
+ * decides otherwise.
+ *
+ * The reader is pinned to the coordinator.  Without that the planner may put a
+ * function scan on the segments, where each of them would read the whole file
+ * and the rows would come back as many times as there are segments.  The writer
+ * cannot say the same -- EXECUTE ON is only accepted for a set-returning
+ * function -- but it does not need to: it is called in a target list with no
+ * FROM clause, which is evaluated on the coordinator, and the query it runs is
+ * dispatched from there like any other.
+ */
+CREATE FUNCTION datalake_parquet_write(path text,
+									   query text,
+									   row_group_size int DEFAULT 0)
+RETURNS bigint AS 'MODULE_PATHNAME' LANGUAGE C STRICT VOLATILE;
+
+REVOKE EXECUTE ON FUNCTION datalake_parquet_write(text, text, int) FROM PUBLIC;
+
+CREATE FUNCTION datalake_parquet_read(path text,
+									  first_row_group int DEFAULT 0,
+									  n_row_groups int DEFAULT 0)
+RETURNS SETOF record AS 'MODULE_PATHNAME' LANGUAGE C STRICT EXECUTE ON COORDINATOR;
+
+REVOKE EXECUTE ON FUNCTION datalake_parquet_read(text, int, int) FROM PUBLIC;

contrib/datalake_fdw/datalake_fdw_test.control

@@ -0,0 +1,24 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+# contrib/datalake_fdw/datalake_fdw_test.control
+
+comment = 'entry points into datalake_fdw internals, for testing'
+default_version = '1.0'
+module_pathname = '$libdir/datalake_fdw'
+relocatable = false
+requires = 'datalake_fdw'

contrib/datalake_fdw/exports.txt

@@ -30,3 +30,10 @@ pg_finfo_iceberg_catalog_fdw_validator
 iceberg_catalog_fdw_validator
 pg_finfo_iceberg_volume_fdw_validator
 iceberg_volume_fdw_validator
+
+# datalake_fdw_test: not part of what this module offers, but a SQL-callable
+# function has to be found by name in the library like any other.
+pg_finfo_datalake_parquet_write
+datalake_parquet_write
+pg_finfo_datalake_parquet_read
+datalake_parquet_read

contrib/datalake_fdw/src/am_iceberg/pg_iceberg_extensible.c

@@ -33,6 +33,7 @@
 #include "access/table.h"
 #include "access/tableam.h"
 #include "am_iceberg/pg_iceberg_ddl.h"
+#include "common/dl_resource.h"
 #include "am_iceberg/pg_iceberg_guc.h"
 #include "am_iceberg/pg_iceberg_options.h"
 #include "am_iceberg/pg_iceberg_reject.h"
@@ -951,6 +952,7 @@ _PG_init(void)
 				 errmsg("datalake_fdw must be loaded via shared_preload_libraries"),
 				 errhint("Add \"datalake_fdw\" to shared_preload_libraries and restart the server.")));
 
+	dl_resource_init();
 	pg_iceberg_define_gucs();
 	pg_iceberg_register_reloptions();
 	DatalakeRegisterMetaEngines();

contrib/datalake_fdw/src/am_iceberg/pg_iceberg_guc.c

@@ -33,6 +33,7 @@
 
 char	   *iceberg_default_catalog;
 char	   *iceberg_default_volume;
+int			iceberg_batch_rows;
 
 void
 pg_iceberg_define_gucs(void)
@@ -64,4 +65,27 @@ pg_iceberg_define_gucs(void)
 							   NULL,
 							   NULL,
 							   NULL);
+
+	/*
+	 * How many rows travel between the executor and a data file at a time.
+	 * Every per-batch cost is paid once per this many rows, and the batch and
+	 * its Arrow copy are held while it is built, so the right value trades
+	 * memory for that -- which depends on how wide the table is, and is why
+	 * this is a setting rather than a constant.
+	 *
+	 * The ceiling is Parquet's default row group length: a batch bigger than
+	 * the unit a file is written in buys nothing.
+	 */
+	DefineCustomIntVariable("iceberg.batch_rows",
+							"Rows per batch exchanged with a lake table's data files.",
+							NULL,
+							&iceberg_batch_rows,
+							16384,
+							1,
+							1024 * 1024,
+							PGC_USERSET,
+							0,
+							NULL,
+							NULL,
+							NULL);
 }

contrib/datalake_fdw/src/am_iceberg/pg_iceberg_guc.h

@@ -31,6 +31,7 @@
 
 extern char *iceberg_default_catalog;
 extern char *iceberg_default_volume;
+extern int	iceberg_batch_rows;
 
 extern void pg_iceberg_define_gucs(void);
 

contrib/datalake_fdw/src/common/dl_err.c

@@ -160,10 +160,20 @@ dl_err_message(DlErrCode code)
 void
 dl_error_report(int elevel, DlErrCode code, const char *prefix)
 {
-	const DlErrorDetail *detail = dl_error_get();
+	/*
+	 * A copy, because reporting consumes the record: what is left otherwise is
+	 * a description of a failure that has already been reported, waiting for
+	 * the next failure with the same code to adopt it -- and the check below
+	 * cannot tell those two apart.  The reset has to happen before the ereport,
+	 * which at ERROR does not come back.
+	 */
+	DlErrorDetail detail_copy = *dl_error_get();
+	const DlErrorDetail *detail = &detail_copy;
 	StringInfoData detail_buf;
 	bool		has_detail;
 
+	dl_error_reset();
+
 	/*
 	 * Detail recorded against a different code belongs to some other failure --
 	 * an implementation that reported this one without recording anything, for

contrib/datalake_fdw/src/common/dl_err.h

@@ -109,10 +109,23 @@ extern const char *dl_err_message(DlErrCode code);
  * when the session asked for log-level detail -- a stack is for whoever is
  * debugging the implementation, not for whoever ran the statement.
  *
- * Detail recorded against a different code is ignored rather than misattributed.
+ * Detail recorded against a different code is ignored rather than misattributed,
+ * and reporting consumes what it used.  Matching on the code alone cannot tell
+ * this failure's detail from an earlier failure's with the same code, so the
+ * record is not left behind for the next one to inherit.
  */
 extern void dl_error_report(int elevel, DlErrCode code, const char *prefix);
 
+/*
+ * An argument that was not what it had to be.  A caller reporting the code
+ * alone would say "invalid parameter" about a call the user never wrote, so
+ * this names the entry point instead -- there is no user error to describe,
+ * only which one of ours was called wrongly.
+ */
+#define DL_ARG_ERROR(operation) \
+	(dl_error_set(DL_ERR_INVALID_OPTION, (operation), NULL, \
+				  "a required argument was missing"), DL_ERR_INVALID_OPTION)
+
 #ifdef __cplusplus
 }
 #endif