Skip to content

datalake_fdw: storage I/O over S3 and pluggable storage backends - #2044

Draft
MisterRaindrop wants to merge 10 commits into
apache:mainfrom
MisterRaindrop:feature/datalake-s3
Draft

datalake_fdw: storage I/O over S3 and pluggable storage backends#2044
MisterRaindrop wants to merge 10 commits into
apache:mainfrom
MisterRaindrop:feature/datalake-s3

Conversation

@MisterRaindrop

@MisterRaindrop MisterRaindrop commented Sep 23, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Draft, and based on #1951 rather than on main: this layer sits under the
Parquet reader and writer that PR adds. The diff therefore still shows
#1951's commits. Once #1951 merges I will rebase, and what is left is the
nine commits of this change alone.

What does this PR do?

contrib/datalake_fdw reads and writes Parquet (#1951), but only on a local
file system. This adds the storage layer underneath: an s3 backend over the
AWS SDK for C++, and a published contract so a backend this extension does not
ship can be added from outside it.

Closes #2009.

Decisions worth a look:

  • A backend answers one question: given a location and its options, which
    arrow::fs::FileSystem reads and writes it. Opening files, listing,
    classifying errors, accounting for memory and translating a URI into a native
    path all stay on this extension's side of the boundary, so a backend is a
    mount function and nothing else.
  • s3 is our own arrow::fs::FileSystem over the AWS SDK, not Arrow's.
    No RPM of Arrow is built with S3 support -- EPEL's and the Arrow project's
    own both set use_s3 0 -- so depending on it would mean asking every user to
    build Arrow. The SDK is a BUILD_ONLY="s3;sts" static build that takes well
    under a minute, links statically, and adds libcurl, OpenSSL and zlib to
    NEEDED and nothing else.
  • Timeouts and retries are set by this module, not inherited: connect 5 s,
    request 300 s, three retries, so a black-holed endpoint answers in about 24
    seconds rather than hanging a session, on every Arrow version.
  • A writer only ever removes what it created. The local backend opens with
    O_EXCL and unlinks that path alone; the S3 one keys cleanup on whether an
    upload is still live rather than on whether the stream is closed, so a
    Close() that fails at the last part still aborts the upload instead of
    leaving parts to be billed for.
  • Credentials are scrubbed at one place. Every value under a key containing
    secret, token or password is remembered per process, and dl_error_set
    -- the single exit every DlErrCode error goes through -- removes it from
    the message. An unexpected C++ exception and a third-party backend's own
    status are covered by construction rather than by remembering to call
    something. A base_path that carries a password in its userinfo is redacted
    the same way, in the message as well as the detail.
  • Which schemes a volume may name comes from the registry, not a list in
    the parser: a backend that registered mine can be written into a
    base_path without this extension being changed.
  • The plug-in contract is versioned three ways: an ABI version, an Arrow
    version, and a fingerprint over the compiler major and _GLIBCXX_USE_CXX11_ABI.
    Registration happens during preload and in any order -- the registering side
    pulls in datalake_fdw through load_external_function, inside a PG_TRY
    so an ereport cannot unwind through the plug-in's C++ frames.
  • Every Arrow allocation still goes through the tracked pool, including a
    backend's: mount is handed a host struct carrying it, and the five
    installed headers give a backend no way to reach Arrow's default pool by
    accident.

Type of Change

  • Bug fix (non-breaking change)
  • New feature (non-breaking change)
  • Breaking change (fix or feature with breaking changes)
  • Documentation update

Test Plan

One body of storage behaviour runs against every backend rather than once per
backend: storage_conformance parameterises a shared SQL file over a URI
prefix and a volume, and runs it for file://, for s3:// and for the test
backend.

  • Integration tests added -- three new categories. storage_local covers
    the facade and the registry: four kinds of bad registration, a duplicate
    scheme, path escapes, a name already in use, a missing file.
    storage_s3 covers the backend against a real service: round trip,
    9 MB through multipart, ListObjectsV2 paging, HeadObject,
    DeleteObject, deleting twice, a missing key, a missing bucket, a wrong
    secret (and that the message does not contain it), path style with an
    endpoint override, and a black-holed endpoint inside 30 s.
    storage_conformance adds the type round trip, field-id projection, a
    row-group range, listing, a rejected overwrite, a write that fails
    halfway leaving nothing behind, a prefix nothing was written under and
    one whose objects were just deleted (both not-found, which is the point:
    object storage and a filesystem disagree and the facade decides), and
    volume resolution: no USAGE, a PUBLIC mapping, a URI outside its
    volume, a volume that does not exist, and a rejected base_path that
    must not echo its own password.
  • Passed make installcheck -- every category, on a three-segment cluster
    with the module preloaded, against MinIO and against SeaweedFS, on
    Arrow 9.0.0 and Arrow 17.0.0.
  • Unit tests added/updated
  • Passed make -C src/test installcheck-cbdb-parallel (not run)

Beyond the suite:

  • The plug-in contract was used from outside. A .cc file that includes
    only the five installed headers and PostgreSQL's server headers compiles to a
    .so that registers a backend; put before datalake_fdw in
    shared_preload_libraries the cluster starts and the scheme works, and
    built with a stale abi_fingerprint the registration is refused with both
    values in the message.
  • Memory was measured, S3 against local. Writing and reading the same
    Parquet file through s3:// costs about 15 MiB of resident memory more than
    through file:// -- and that difference stays flat as the object grows
    (7.9 MiB for a 2 MB file, 14.8 MiB for a 149 MB one), so neither the response
    body nor the multipart buffer accumulates.
  • The credential chain was exercised. With no user mapping the same
    statement fails; with AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY in the
    postmaster's environment it succeeds.
  • Nothing new is exported. nm -D lists the PostgreSQL entry points, one
    registration function and the test extension's UDFs; no Arrow and no AWS
    symbols, on all three build variants.

Impact

Dependencies: an optional build dependency on the AWS SDK for C++. Without
it the extension builds as before and says so, and opening an s3:// location
reports that it was built without it; naming a prefix that has no SDK in it is
an error rather than a silent fallback. The module is off by default and not in
the RPM, so packaging is unchanged.

CI builds the SDK from source once per distribution and architecture and caches
it, starts SeaweedFS for the s3 cases, and now takes Arrow from the Arrow
project's own repository on Rocky 9 and 10 as well, pinned -- 17.0.0 and
21.0.0, so a version change in EPEL cannot arrive without a commit.

What that costs the ic-datalake-fdw job, measured on all three legs: building
the SDK takes 175-228 s on a cache miss and the cache restores in 1 s (it is
4 MB); SeaweedFS is up 6 s after the step starts; the s3 cases add about 35 s,
of which 22 s is one deliberate connect timeout.

User-facing changes: a volume's base_path now accepts file:// as well
as s3://, and the scheme it rejects is named against what is registered
rather than against a fixed list. Three iceberg_am_reject DETAIL lines and
two ERROR lines change wording, the latter because they used to quote back a
URI that can carry a password.

Checklist

contrib/datalake_fdw/README.md documents volumes, where credentials come
from, building with S3 support, and writing a backend -- the example in it is
compiled against the installed headers rather than written out by hand. The
test extension's new functions name a path on the server's file system and are
revoked from PUBLIC like the two that were already there.

Additional Context

Known limits, deliberately rather than by oversight:

  • A file:// volume must be the same directory on every host. Nothing
    checks it; the README says so.
  • Between checking that a name is free and creating it, another writer can
    take it.     Iceberg's file names are unique by construction, so it does not
    arise there.
  • An upload abandoned by a crashed backend leaves its parts behind. A
    bucket lifecycle rule that expires incomplete multipart uploads is the usual
    answer; the README says so.
  • About 15 MiB per backend process is outside gp_vmem_protect_limit --
    the SDK client, its connection and one part buffer are allocated by the SDK
    rather than through the tracked pool.
  • The fingerprint catches the mismatches that occur in practice, not every
    possible one.
  • No HDFS. The scheme is gone from the parser rather than half-supported;
    it returns as a backend when someone needs it.

MisterRaindrop and others added 3 commits September 3, 2026 18:15
@MisterRaindrop
datalake_fdw: read and write Parquet through Arrow
777ba63 
The format layer had an interface and no implementation.  This is the
Parquet one, and the conversions either side of it: PostgreSQL tuples
into an Arrow batch, and an Arrow batch back into Datums.

Arrow rather than libparquet alone, because libparquet is written in
terms of Arrow's types, so linking one links the other.  It is a build
dependency now, found with pkg-config -- and whatever -std= its .pc file
asks for is filtered out, because those flags land after CXXFLAGS and
the Arrow project's own packages say -std=c++11.

A fragment is a range of row groups rather than a whole file, so several
segments can read one large file.  Reading is single-threaded: a worker
thread that fails has no way to report it through PostgreSQL.  A reader
and a writer hold a descriptor and memory from Arrow's allocator, which
transaction abort does not reclaim, so both register with the resource
owner -- the shape PAX uses in comm/pax_resource.cc.

Types: bool, the four integers, both floats, text, varchar, char, bytea,
date, timestamp, timestamptz.  Anything else is refused by name.  The
30-year epoch shift is range checked in both directions: PostgreSQL's
range runs past what Arrow holds as microseconds from 1970, and a round
trip through two unchecked halves would agree with itself.

datalake_fdw_test is a second extension in the same library with the two
functions this can be run with from SQL.  Its regression case round-trips
every supported type and checks that reading row groups separately gives
back what reading the whole file does.

Arrow 9.0.0 (EPEL 9) builds and passes; 17.0.0 (Rocky 10, gcc 14) and
17.0.0 and 21.0.0 (Rocky 8, gcc 8) compile without warnings.  What is
written here reads back correctly in pyarrow 21, which is not the
implementation that wrote it.
@MisterRaindrop
datalake_fdw: address the Parquet review, field ids and tracked memory
c75f6e9 
Review of apache#1951, plus what the same pass turned up.  The framework is what
this settles; type coverage beyond it is left to separate issues.

Memory.  Every Arrow allocation now goes through one pool
(format/arrow_memory_pool.cpp) that reserves with the vmem tracker before
allocating and releases after freeing, so statement_mem, the resource group
and gp_vmem_protect_limit see Arrow's memory as they see palloc's.  The
reserve runs under HOLD_INTERRUPTS, because the tracker can elog(ERROR) on
its way to saying no and a longjmp out of Arrow's C++ frames is undefined
behaviour; pre_buffer is switched off by name, because from Arrow 13 it
defaults on and allocates from I/O threads the tracker cannot see.  All
five pool sites use it, the two hidden defaults included.  A refusal is
DL_ERR_OUT_OF_MEMORY, ERRCODE_OUT_OF_MEMORY.

Files.  The writer creates its file with O_CREAT|O_EXCL and hands the
descriptor to Arrow, so a failed write can no longer truncate and then
delete a file that was already at the path; every failure path goes
through parquet_discard, which only ever removes what this writer created.
Compression names are lower-cased and asked of Arrow in three steps, so
the message says whether Arrow, Parquet or this build is what refused.

Columns.  A table's columns are matched to a file's by Iceberg field id:
the writer stamps PARQUET:field_id into the Parquet schema, ProjectionSet
names field ids in output order (ABI 2), a field the file lacks reads as a
null-typed column, and an id-less column can never be matched.  The
reader also accepts the promotions the spec allows, int32 as bigint and
float as double precision.  Dropped attributes are skipped in the schema,
the batch builder and the writer.

Types.  time and uuid are added; large_utf8 and large_binary are read;
INT96 timestamps are coerced to microseconds.  Dictionary-encoded columns
are refused rather than decoded as their indexes.  Both sides require a
UTF8 database, and the reader verifies every string before it becomes a
text.  varchar(n) and char(n) are refused, with numeric and everything
else the format cannot store, at CREATE TABLE, through one function the
DDL hook and the writer both ask (format/format_types.h); LIKE is refused
because its columns are resolved after the hook has run.

Smaller: dl_resource.c uses lib/ilist.h; the format name is compared
without regard to case like every other option value; the test functions
gain compression and field_ids arguments and check PG_NARGS() so that a
stale extension definition errors instead of crashing.

Left for later, each with an issue: NUMERIC, millisecond and nanosecond
timestamp columns, name mapping for files without field ids, and the CI
build image.
@MisterRaindrop
Merge branch 'main' into feature/datalake-parquet
d4a23ac
@MisterRaindrop
datalake_fdw: add a pluggable storage layer
9e8de38 
Where a lake table's files live was decided by a switch on the URI
scheme, and only s3 and hdfs were written into it.  This replaces that
with a contract: a backend answers one question -- given a location and
its options, which arrow::fs::FileSystem reads and writes it -- and
everything else stays on this extension's side of the boundary.

What lands here:

- The contract, in five headers that "make install" puts under
  $(includedir_server)/extension/datalake_fdw/, so a backend can be
  built outside this tree against nothing else.  It carries an ABI
  version, the Arrow version, and a fingerprint over the compiler major
  and _GLIBCXX_USE_CXX11_ABI, because shared_ptr and arrow::Result
  cross the boundary by value.
- A registry keyed by URI scheme, found through a rendezvous variable.
  Registration happens during preload and in any order: the registering
  side pulls in datalake_fdw through load_external_function, inside a
  PG_TRY so an ereport cannot unwind through a plug-in's C++ frames.
  Each process initializes a backend at its first mount and registers
  the matching finalizer then, because on_exit_reset() clears what a
  postmaster child inherits.
- The facade over that filesystem: opening, listing, deleting,
  turning a URI into a native path, and classifying failures into
  DlErrCode.  Backends never re-parse a URI, and an empty listing is
  ruled not-found here rather than by each backend, because object
  storage and a filesystem disagree about what an empty prefix means.
- A file backend for shared mounts, which creates with O_EXCL and, if
  it has to give up, removes the file this writer created and nothing
  else.
- Credential scrubbing in one place.  Every value under a key
  containing secret, token or password is remembered per process, and
  dl_error_set -- the single exit every DlErrCode error goes through --
  removes it from the message, so an unexpected C++ exception and a
  third-party backend's own status are covered by construction.
- A dltest backend, so the registration contract can be tested from
  inside this module.

The parser now accepts file:// as well as s3://, and names what it
rejects against the registry rather than a fixed list, so a scheme a
plug-in registered can be written into a base_path.  Three DETAIL lines
in iceberg_am_reject change with it.

The s3 backend is adapted to the new contract here and rewritten in the
commit that follows.
@MisterRaindrop
datalake_fdw: read and write s3 through the AWS SDK
4a3ff10 
The s3 backend was a stub.  This makes it an arrow::fs::FileSystem of
our own over the AWS SDK for C++, rather than Arrow's S3FileSystem: no
RPM of Arrow is built with S3 support -- EPEL's and the Arrow project's
own both set use_s3 0 -- so depending on it would mean asking every
user to build Arrow.

A synchronous S3Client, so a cancelled query cannot leave a callback
holding a backend's memory.  Reads go through PreallocatedStreamBuf
straight into a tracked buffer rather than the SDK's default
stringstream.  Writes buffer 8 MiB and begin a multipart upload only
when they exceed it, so a small file is one PutObject.  Cleanup depends
on whether an upload is still live rather than on whether the stream
was closed, so a Close() that fails at the last part still aborts
instead of leaving parts to be billed for.

Connect timeout 5 s, request timeout 300 s, three retries, set here
rather than inherited, so a black-holed endpoint answers in about 24
seconds on every Arrow version.

The SDK is found by prefix -- the one AWS_SDK_PREFIX names, or the
usual places a hand-built one lands.  Without it the module still
builds and says so, and opening an s3:// location reports that it was
left out; a prefix with no SDK in it is an error rather than a silent
fallback.  It links statically and adds libcurl, OpenSSL and zlib to
NEEDED and nothing else.
@MisterRaindrop
datalake_fdw: read and write Parquet wherever the volume is
0748365 
The Parquet reader and writer opened paths through the local file
system directly.  They now open through the storage facade, so a
fragment names a filesystem and a path relative to its mount, and the
same reader works against a volume on s3 as on a shared mount.

"Create only" moves with them: the writer asks the facade, which is the
one place that checks a name is free, and giving up is the stream's
Abort rather than a delete by path -- after a failed write the name may
already belong to somebody else.

A volume resolves to a location plus credentials in one place, checking
USAGE on the server and reading the user mapping, the PUBLIC mapping,
or neither.
@MisterRaindrop
datalake_fdw: run one body of storage behaviour against every backend
0293d0b 
Three backends that must behave identically are worth one body of
tests, not three that drift.  A shared SQL file is parameterised over a
URI prefix and a volume and run for file://, for s3:// and for the test
backend: the type round trip, field-id projection, a row-group range,
listing, a rejected overwrite, and a write that fails halfway leaving
nothing behind.

Volume resolution gets its own case, on a file volume so it runs
everywhere: no USAGE, a PUBLIC mapping, a URI outside its volume, and a
volume that does not exist.
@MisterRaindrop
datalake_fdw: document the storage layer
efaa71d 
Volumes and their options, where credentials come from and in what
order, what a file:// volume requires of the filesystem, how to build
with S3 support, and how to write a backend -- the example is compiled
against the installed headers rather than written out by hand.  Plus
the limits that are accepted rather than fixed.
@MisterRaindrop
datalake_fdw: close what two review rounds and a manual audit found
b76f2fb 
Two independent adversarial review rounds, then a read of the whole
diff against this change's own invariants.  The defects worth naming:

- A multipart upload was abandoned rather than aborted.  Close() marked
  the stream closed before uploading the last part, so the Abort() that
  followed a failure did nothing and the parts stayed, billable.
  Cleanup now depends on the upload being live, and the destructor does
  the same.
- A local file could outlive the writer that created it: the guard was
  released one statement before the wrapper took ownership, and the
  wrapper had no destructor, so an exception rather than an error code
  left the file behind.
- Two paths skipped credential scrubbing -- an unexpected exception's
  what(), and the format layer's own status mapping.  Scrubbing moved
  to dl_error_set, which every error goes through.
- A bare "404" was matched inside the whole status text, so an error
  about a directory named 404 read as not-found.  Only phrases a
  filesystem writes about itself are matched now, and the README said
  three others that never were.
- ReadAt trusted the length the service returned; a paginated listing
  whose continuation token stopped advancing looped forever; half a
  credential pair silently fell back to the host's own identity; a
  bucket that does not exist looked like a directory.
- The validator quoted a rejected base_path back verbatim in its
  message while the detail beside it had just stopped doing so, so
  s3://user:password@host/path wrote the password to the server log.

And in the tests, which were also under review:

- A loop of the shape "SELECT count(*) FROM (SELECT f(i) FROM
  generate_series(...)) w" never called f at all -- the target list is
  unreferenced, so it is optimised away while the count still looks
  right.  The pagination case wrote nothing for 1100 rows and passed.
  Every such loop now reads the value the function returned.
- The redaction case asserted that a message did not contain a secret
  the service never echoes, so it passed with redaction removed
  entirely.  It uses the bucket name as the secret now.
- The paginated listing case had no expected output at all, so the one
  way CI is meant to run it could only fail.
- Nothing checked that an empty prefix reads as not found, which is a
  rule this change introduced.

Measured while auditing: reading and writing through s3:// costs a
backend about 15 MiB of resident memory more than the same file
locally, and that stays flat as the object grows -- 7.9 MiB for a 2 MB
file, 14.8 MiB for a 149 MB one -- so neither the response body nor the
part buffer accumulates.  It is outside gp_vmem_protect_limit, which
the README now says.
@MisterRaindrop
ci: build the AWS SDK and run datalake_fdw's s3 cases
dbd0f51 
The s3 half of contrib/datalake_fdw's regression had nothing to run
against.  Three things were missing.

The AWS SDK for C++, which no distribution packages: built from source
with BUILD_ONLY="s3;sts", and cached under the version, the
distribution and the architecture, since nothing else changes it.  A
miss costs about three minutes; the cache is 4 MB and restores in one
second.  Only the build dependencies the image lacks are installed,
asked for by capability -- naming a package it already has makes dnf
try to upgrade it, and on Rocky 10 the newest libcurl-devel wants a
libcurl no enabled repository carries.

A service that speaks S3: SeaweedFS, one static binary, pinned, and
started with an identity file -- without one it accepts any
credentials, and the case that asserts a wrong secret is refused would
pass by not being tested.  Readiness waits on the S3 port itself, which
begins listening seconds after the master elects itself.

Its coordinates, which "su - gpadmin" drops along with the rest of the
environment, so they are named on the command line.  Every other test
entry leaves that empty, and a leg where the service did not start
skips the s3 cases rather than failing them.

Arrow on Rocky 9 and 10 now comes from the Arrow project's repository
pinned to 17.0.0 and 21.0.0, as it already did on Rocky 8.  EPEL's
moves when EPEL does, and Arrow is the library this extension's ABI is
shared with.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

No reviews

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

datalake_fdw: storage I/O over object storage (S3)

1 participant

@MisterRaindrop