feat(format-table): support reading and writing format tables - #222
feat(format-table): support reading and writing format tables#222SteNicholas wants to merge 1 commit into
Conversation
b975f5e to
595a9ee
Compare
fc0be5d to
e654101
Compare
zjw1111
left a comment
There was a problem hiding this comment.
Thanks for the contribution. I found two correctness issues in the write and commit paths; details are inline.
3ded34f to
364136b
Compare
bb281a5 to
42445a6
Compare
zjw1111
left a comment
There was a problem hiding this comment.
One API design concern about the catalog hierarchy.
ec7d57e to
1a84946
Compare
A format table is a directory of data files of one format with no snapshots and no manifests, laid out like a standard Hive table. Only `parquet` and `orc` are supported. `Catalog::GetFormatTable()` loads one through `Catalog::LoadFormatTable()`, a protected virtual the file system catalog overrides to say its metadata lives under the table location, and the REST catalog to take the location and the schema from one response. A caller holding a table path keeps the interface it uses for every other table: `TableScan::Create()`, `TableRead::Create()`, `FileStoreWrite::Create()` and `FileStoreCommit::Create()` recognise a format table from the schema under the path and dispatch to it, the way Java Paimon serves both kinds through one `ReadBuilder` and one `BatchWriteBuilder`. `FormatTable` is the only format table type in the public API: the scan, read, write and commit classes, the split and the commit message are implementation details under `src`, and a caller sees them as `Plan`, `Split` and `CommitMessage`. `TableScan` gains `ListPartitions()`, which a format table answers by listing directories. One `FormatTableLoader` answers the "is this a format table" question for all four entry points and hands back the schema it read, so the schema file is read once. Anything a context carries that a format table cannot honour is refused by name rather than quietly dropped. The scan discovers partitions by descending the directory layout - `key=value` by default, the bare value under `format-table.partition-path-only-value` - and packs a partition's files into splits of about `source.split.target-size`, each holding whole files. The read projects the table's columns, pushes a predicate into the file readers and applies it exactly on request, and rebuilds the partition columns from the directory names, since the data files themselves do not carry them. Partition completion, the predicate split and the read-schema order all come from `FieldMappingReader`, and a data file is opened through `DataFileReaderFactory`, which the managed table path uses too, so prefetch and the caches a `ReadContext` carries apply here as well. The write is two-phase: it stages each file under `_temporary/.tmp.<uuid>` beside where it will be published, the layout Java's `RenamingTwoPhaseOutputStream` stages under, and the commit renames them into place. An overwriting commit first clears the partitions it writes to, or the static partition it was given. Options are read through `CoreOptions`, so a default lives in one place. A split and a commit message both come back through an interface that takes the base type, so each is checked for path containment, visibility and partition binding before anything is read, renamed or deleted. Neither has a serialized form: a format table's plan has no cross-runtime encoding, so `Split::Serialize()` and `CommitMessage::Serialize()` refuse them and a plan is read within the process that made it. - `Catalog::GetTable()` and `Table::Create()` refuse a format table, and every other table type this library does not implement, instead of opening it as a managed table and looking for snapshots it never had. - Every catalog validates a new table's schema through one entry point, so none accepts a table another could not open; `RestCatalog::CreateTable()` runs it now too. - `PartitionPathUtils::GeneratePartitionPath()` takes the value-only layout as a parameter and refuses a partition value that cannot name a directory. The managed table path goes through it too. - `LazyConcatBatchReader` opens one file of a split at a time, names the file in every failure and keeps answering with the failure it stopped at, as `BatchReader` requires. `docs/source/user_guide/format_table.rst` describes the layout, the options, and what is still missing against Java, including `dynamic-partition-overwrite` and partition filters beyond equality.
1a84946 to
ee555ee
Compare
| PAIMON_RETURN_NOT_OK(ValidateGenericTableSchema(schema)); | ||
| // At creation the schema's own options are the only ones there are, and `file-system` is | ||
| // resolved from them like every other option. | ||
| return ValidateFormatTableSchema(schema, schema.Options(), /*file_system=*/nullptr); |
There was a problem hiding this comment.
Should ValidateNewTableSchema accept the caller’s already resolved FileSystem, so that we can pass it here instead of nullptr? SchemaManager already has file_system_. This is harmless today because validation performs no filesystem I/O, but passing the actual instance would better match the API contract and avoid resolving a different filesystem in the future.
| /// prefix: `<location>/../victim` starts with the location and still resolves outside it. | ||
| /// Only the path text is checked, so a symbolic link pointing out of the table is not caught. | ||
| static Status ValidatePathUnderLocation(const std::string& path, const std::string& location, | ||
| const std::string& what); |
There was a problem hiding this comment.
Could we normalize the location and candidate path before performing the containment check? The current string comparison may reject equivalent local paths, for example a table location /tmp/table and a file path file:///tmp/table/data.parquet.
Path::ToString() alone would not fully address this, since it normalizes file:///... to file:/... but does not make it equivalent to a scheme-less local path. Perhaps we could parse both paths with PathUtil::ToPath(), compare the scheme and authority separately—treating an empty scheme and file as equivalent for local paths—and perform the component check against the normalized Path::path. A regression test covering mixed local-path and file: URI forms would also be helpful.
|
|
||
| /// Renames every written file into place, first clearing what it replaces when the commit | ||
| /// overwrites. | ||
| Status Commit(const std::vector<FormatCommitMessage>& commit_messages); |
There was a problem hiding this comment.
Could we either support or explicitly reject dynamic-partition-overwrite=false for format tables? Currently, overwrite derives the directories to clear only from commit messages, so it always behaves like dynamic partition overwrite and clears nothing when the message list is empty. Java replaces the whole table when this option is false, and always does so for unpartitioned tables. If this behavior is out of scope, failing fast may be safer than silently leaving stale data.
|
|
||
| Status Commit(const std::vector<std::shared_ptr<CommitMessage>>& commit_messages, | ||
| int64_t commit_identifier = BATCH_WRITE_COMMIT_IDENTIFIER, | ||
| std::optional<int64_t> watermark = std::nullopt) override; |
There was a problem hiding this comment.
Please avoid using default parameters in production code.
| bool enable_predicate_filter) { | ||
| return CreateInternal(table, projection, pool, predicate, enable_predicate_filter, | ||
| /*read_context=*/nullptr); | ||
| } |
There was a problem hiding this comment.
Is this function intended as a test-only interface, or is it also used in production code? If it is only for tests, could we move it to private, rename it to something like TEST_Create, or have the tests call FormatTableRead::Create directly from read_context instead?
The main concern is that the parameters of CreateInternal look a bit unusual right now. For example, enable_predicate_filter and predicate already exist in read_context, but currently some of this information is passed separately while some is taken from read_context, which makes the interface feel inconsistent.
| "different version of the file", | ||
| file.file_size, status.GetLen())); | ||
| } | ||
| // Opened through the same component the managed table path opens a data file with, |
There was a problem hiding this comment.
There may be a potential performance hotspot here. In the read path, both StarRocks and DuckDB previously observed that, in small-file scenarios, calling OSS open after first fetching the file length could become a noticeable hotspot. They optimized this by opening the file directly using the file size from metadata, which helped reduce the open overhead (pr #189 ). It may be worth leaving a TODO here so we can optimize this later if it turns out to be a hotspot in practice.
| value = std::move(key_value->second); | ||
| } | ||
| auto filter_iter = partition_filter_.find(partition_key); | ||
| if (filter_iter != partition_filter_.end() && filter_iter->second != value) { |
There was a problem hiding this comment.
Could we consider evaluating partition filters on a typed BinaryRow instead of comparing raw path strings?
PAIMON_ASSIGN_OR_RAISE(
BinaryRow partition,
partition_computer_->ToBinaryRow(partition_spec));
PAIMON_ASSIGN_OR_RAISE(
bool matched,
partition_filter_->Test(partition_schema_, partition));This would align the behavior with Java and the regular table scan path, particularly for typed values and null partition semantics, while also making richer partition predicates easier to support later.
|
|
||
| /// Renders a partition column as the text a partition directory is named with. | ||
| Result<std::shared_ptr<arrow::StringArray>> RenderPartitionColumnAsText( | ||
| const std::shared_ptr<arrow::Array>& column, const std::string& field_name, |
There was a problem hiding this comment.
This check may not be necessary. The original write path in paimon-cpp does not perform this validation either, and it could introduce some performance overhead. Also, even if this value were written incorrectly, it should not have any practical impact, since the paritition field in the file is not read.
| PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*data_struct, &c_data_array)); | ||
| PAIMON_RETURN_NOT_OK(file_iter->second.writer->AddBatch(&c_data_array)); | ||
| file_iter->second.record_count += data_struct->length(); | ||
|
|
There was a problem hiding this comment.
Could we abort the writer or mark the write as terminal when AddBatch or ReachTargetSize fails? The writer may already be partially modified, so allowing subsequent Write or PrepareCommit calls could produce an incomplete file. This would also align the failure handling with the existing RollingFileWriter and Java implementation.
| ASSERT_FALSE(created.ok()); | ||
| ASSERT_NE(std::string::npos, created.status().ToString().find("cannot be partitioned")) | ||
| << created.status().ToString(); | ||
|
|
zjw1111
left a comment
There was a problem hiding this comment.
Thanks for this - the format table write/commit path lines up closely with Java Paimon ({prefix}{uuid}-{n}.{ext}, the _temporary/.tmp.<uuid> staging layout, the validate-then-clear-then-rename ordering), and the option refusals are well documented.
A few non-blocking cleanups below, plus one design note that is explicitly not for this PR.
| static Result<FormatFileNaming> Create(const std::string& extension, const std::string& prefix); | ||
|
|
||
| FormatFileNaming() = default; |
There was a problem hiding this comment.
Create() validates four things here - non-empty extension, no path separator or .. in either extension or prefix, prefix not hidden by the _/. convention, and successful UUID generation - so construction can genuinely fail. But the public FormatFileNaming() = default; on line 49 lets a caller bypass all of it: a default-constructed instance has an empty uuid_ and extension_, and NextFileName() would then produce data--0.. It seems to exist only so that FormatTableWrite::Impl can hold it as a value member (format_table_write.cpp:184, assigned later at :297).
Could you change Create() to return Result<std::unique_ptr<FormatFileNaming>> and drop the public default constructor? Impl::naming would become a std::unique_ptr, and the invariant would stay inside the type. That also matches the static Create() + private constructor rule in docs/code-style.md:187.
| /// A file is staged under `_temporary/.tmp.{uuid}` beside where it will end up and takes its real | ||
| /// name only on commit, as Java Paimon's `RenamingTwoPhaseOutputStream` does. Both the directory | ||
| /// and the name are hidden, which is the Hive-style convention for output that is not committed | ||
| /// table data and is what a scan of this table skips. |
There was a problem hiding this comment.
A note on the design rather than something to change here.
Mirroring RenamingTwoPhaseOutputStream is the right starting point, but it ties publishing to a rename. On an object store a rename is neither atomic nor cheap - it is a server-side copy followed by a delete - so a commit that publishes many files effectively pays for the data twice, and a failure part-way through the rename loop cannot be undone atomically (the rollback in format_table_commit.cpp:292-308 is best-effort for exactly this reason).
Java has a second implementation for this case: MultiPartUploadTwoPhaseOutputStream (paimon-common/src/main/java/org/apache/paimon/fs/MultiPartUploadTwoPhaseOutputStream.java, with the OssTwoPhaseOutputStream / S3TwoPhaseOutputStream / JindoTwoPhaseOutputStream subclasses). There the data is uploaded straight to its final key and only CompleteMultipartUpload is deferred to commit, so there is no copy and the publish is a single atomic call.
This PR is already very large, so please don't change it here. Could you just leave a TODO in this comment block noting that only the rename-based semantics are supported today and pointing at MultiPartUploadTwoPhaseOutputStream as the follow-up? The object-store path can then be optimised in a later PR.
| PAIMON_ASSIGN_OR_RAISE(bool is_system_table, identifier.IsSystemTable()); | ||
| if (is_system_table || CatalogUtils::IsSystemDatabase(identifier.GetDatabaseName())) { | ||
| return Status::Invalid(fmt::format("{} is a system table, so it cannot be a format table", | ||
| identifier.GetFullName())); | ||
| } |
There was a problem hiding this comment.
CatalogUtils::CheckNotSystemTable (src/paimon/core/catalog/catalog_utils.cpp:55) already wraps exactly this pair of checks, and its comment spells out why the ordering matters: "The system database is checked first so that an identifier of 'sys' is rejected without being parsed as a table name."
This evaluates identifier.IsSystemTable() first, which is the order that comment warns against. IsSystemTable() goes through SplitTableName() (identifier.cpp:117-155), which returns Status::Invalid("Invalid table name: ...") for a malformed name - so sys.<malformed> surfaces a parse error instead of the system-table refusal. The same pattern appears a few lines up at :384.
Could you call CatalogUtils::CheckNotSystemTable(identifier, "load format table") here instead? If the format-table specific wording is worth keeping, extracting an IsSystemTableIdentifier() helper into CatalogUtils that preserves the database-first order would let both call sites share one implementation.
| return Status::Invalid("Partition value '" + value + | ||
| "' cannot be used as a partition path component."); |
There was a problem hiding this comment.
docs/code-style.md:308-310 (String Formatting) asks for fmt::format() rather than + concatenation, and every other Status::Invalid added in this PR already follows it. Could you switch this one too?
return Status::Invalid(
fmt::format("Partition value '{}' cannot be used as a partition path component.", value));|
|
||
| ~FormatCommitMessage() override = default; | ||
|
|
||
| std::string ToString() const; |
There was a problem hiding this comment.
ToString() is declared here but defined in format_table_commit.cpp:134, and there is no format_commit_message.cpp in this directory. Any translation unit that includes this header without also linking format_table_commit.cpp - a future unit test covering the commit message on its own, for instance - would hit an undefined symbol.
Could you define it inline in the header? The body is a single fmt::format call, so it would need fmt/format.h here, which seems a fair trade for keeping the header's contract and its implementation together.
zjw1111
left a comment
There was a problem hiding this comment.
One follow-up on the documentation, after a closer read of the symbol names in this file.
| ``RecordBatch::SetPartition()``, every row is checked against that declaration, and a batch | ||
| mixing partitions is refused. Java routes row by row, so one write call there may land in any | ||
| number of partitions; | ||
| * a write takes its partition from ``RecordBatch::SetPartition()`` rather than from the rows, so |
There was a problem hiding this comment.
Small naming fix: SetPartition() is declared on RecordBatchBuilder, not on RecordBatch - include/paimon/record_batch.h:126 has
RecordBatchBuilder& SetPartition(const std::map<std::string, std::string>& data);inside class PAIMON_EXPORT RecordBatchBuilder (line 97), and class PAIMON_EXPORT RecordBatch (line 39) has no such method. Could you change both mentions here to RecordBatchBuilder::SetPartition()? A reader following the first bullet would otherwise look for it on the wrong type.
Every other symbol this page names checks out, for what it is worth - ReadContextBuilder::SetReadSchema, WriteContextBuilder::WithWriteSchema / WithWriteId, CommitContextBuilder::IgnoreEmptyCommit, TableRead::CreateCountReader and TableScan::ListPartitions all match their declarations.
Purpose
Linked issue: close #170
A format table is a directory of data files of one format with no snapshots and no manifests, laid
out like a standard Hive table. A table is one when its
typeoption isformat-table;file.formatthen names the format of every file in it,parquetororchere. Such a directoryrecords no row identity, so the table only accepts inserts.
A caller reaches one through the interfaces it already uses.
Catalog::GetFormatTable()loadsthe table, and
TableScan::Create(),TableRead::Create(),FileStoreWrite::Create()andFileStoreCommit::Create()each recognise a format table from the schema under the table path anddispatch to it, the way Java Paimon serves both kinds through one
ReadBuilderand oneBatchWriteBuilder.FormatTableis the only format-table type in the public API; the scan, read,write and commit classes, the split and the commit message live under
src, and a caller sees themas
Plan,SplitandCommitMessage. OneFormatTableLoaderanswers the "is this a formattable" question for all four entry points and hands back the schema it read, so the schema file is
read once rather than twice. Anything a context carries that a format table cannot honour is
refused by name rather than quietly dropped.
key=valuebydefault or by the value alone under
format-table.partition-path-only-value; the one namedpartition.default-namestands for a null value. The scan descends that layout, unescapes thenames back into values, keeps the partitions an equality filter accepts, and can return them
alone through the new
TableScan::ListPartitions(). It then lists a partition's files at anydepth below it, skipping hidden and staging names, and packs them into splits of about
source.split.target-size, each file counted as at leastsource.split.open-file-cost. A splitholds whole files:
parquetandorcrecord where their own row groups and stripes begin, so abyte range of one would tell a reader nothing.
rather than from the files, pushes the predicate into the file readers and applies it exactly
under
enable_predicate_filter. Partition completion, the predicate split and the read-schemaorder all come from
FieldMappingReader, and a data file is opened through the newDataFileReaderFactory, whichAbstractSplitReaduses too, so prefetch and the caches aReadContextcarries apply here as well. A split's files are opened as they are reached, so alarge partition holds one open file, not one per file.
target-file-row-numortarget-file-size, and stages each under_temporary/.tmp.<uuid>beside where it will be published, the layout Java's
RenamingTwoPhaseOutputStreamstages under.The commit renames them into place; with
overwriteit first clears the partitions its messagesname, or the one a static partition spec names. A file that cannot be closed ends the write,
since its rows can never be published and publishing the others would lose them silently.
Abort()removes what is still staged.Catalog::GetFormatTable()calls a protected virtualCatalog::LoadFormatTable(). The default reads the location and the schema through the virtualsevery catalog has;
FileSystemCatalogoverrides it to say its metadata lives under the tablelocation, which is what tells a
schemaorbranchdirectory there from a partition of the samename, and
RestCatalogto take both from oneGetTableresponse instead of two that coulddisagree.
catalog accepts a table another could not open, and runs them again when the table is opened,
since a schema can reach a catalog without having passed through creation here. Options are read
through
CoreOptions, so a default lives in one place.the base type, so each is checked for path containment, visibility and partition binding before
anything is read, renamed or deleted. One naming a file outside the table, a hidden or staged
file, this table's own metadata, or a partition it does not belong to is refused up front.
Tests
102 cases in four files under
src/paimon/core/table/format/, plus six existing test filesupdated.
format_table_test.cpp(83 cases) carries the end-to-end coverage:directory, a null partition value, batches of one partition sharing one file, a zero-row batch
writing nothing, and the compression suffix in a file's name.
the schema's, a branch and a caller-held schema honoured,
typenever decided by an option, anoverwrite through
FileStoreCommit::Overwrite(), and prefetch and the read context's cachereaching the format reader.
FileStoreCommitcall about snapshots, a compaction, a write id, astreaming or real-time context, a projected read schema, a scan predicate, non-
INSERTrows, abatch whose rows disagree with the partition it declares, and a split or commit message naming a
file outside the table, a hidden or staged file, this table's metadata, or a partition it does
not belong to.
GetPos(),Flush()orClose()ends thewrite rather than leaving a writer that is no longer there, an overwrite of a nested target
replacing the whole partition, and a commit that fails part way taking its published files back.
against the target size, a file larger than the target staying one split, and data files found in
subdirectories.
Split::Serialize()andCommitMessage::Serialize()both refusing.The other three files cover one component each:
format_file_listing_test.cpp(5),format_file_naming_test.cpp(7) andlazy_concat_batch_reader_test.cpp(7), the last includingthat a failure is terminal and never reaches the next file.
Updated:
core/core_options_test.cpp(the format table options and the file compression chain),core/utils/partition_path_utils_test.cpp,core/table/table_test.cpp,core/catalog/file_system_catalog_test.cpp,rest/rest_catalog_test.cppandrest/rest_messages_test.cpp.No integration test under
test/inte/and no benchmark: everything here is exercised through thepublic API from unit tests.
API and Format
One new public header,
include/paimon/table/format/format_table.h:FormatTable, builteither from a table path whose schema it reads itself or from a schema a catalog already holds, and
the
Formatenum (PARQUET,ORC).paimon/api.hexports it.Two existing public interfaces grow:
Cataloggains the publicGetFormatTable()and the protected virtualLoadFormatTable()described above.
Catalog::GetTable()andTable::Create()now refuse a format table, and everyother table type this library does not implement, instead of opening it as a managed table and
looking for snapshots it never had.
TableScangainsListPartitions(), a virtual whose default returnsNotImplemented. Only theformat table path answers it today, where partitions are the directories a scan descends.
defs.haddsTYPE,FORMAT_TABLE_FILE_COMPRESSION,FORMAT_TABLE_PARTITION_PATH_ONLY_VALUE,METASTORE_PARTITIONED_TABLEandFILE_SUFFIX_INCLUDE_COMPRESSION, andCoreOptionsreads them.The rest stays behind the headers:
PartitionPathUtils::GeneratePartitionPath()takes thevalue-only layout as a parameter and refuses a partition value that cannot name a directory, which
the managed table path goes through too;
RestCatalog::CreateTable()validates the schema on theoptions it will send;
AbstractSplitReadopens its data files through the sharedDataFileReaderFactory; andPredicateBatchReaderimports its batch throughPAIMON_ASSIGN_OR_RAISE_FROM_ARROW.Storage format: unchanged. Data files are plain
parquetororcwith nothingpaimon-specific in them, and no manifest, snapshot or index format is touched. A format table's
plan and commit messages have no serialized form at all:
Split::Serialize()andCommitMessage::Serialize()refuse them, since there is no cross-runtime encoding for either and aplan is read within the process that made it.
Documentation
docs/source/user_guide/format_table.rstis the new user-facing page. The part worth reading firstis "Current limits", which lists what Java Paimon's format tables do and this implementation
does not yet: the
csv,json,textandmosaicformats, byte-range splits,metastore.partitioned-tableand the Hive partition sync with it, partition filters beyondequality,
scan.ignore-corrupt-files/scan.ignore-lost-files,dynamic-partition-overwrite,column default values, a table partitioned by every one of its columns, and
TIMESTAMP/DECIMAL/FLOAT/DOUBLEpartition columns. The rest of the page covers the directory layoutand both partition layouts, the read and write path and which objects may be shared between
threads, the two-phase write and what
Abort()does and does not undo, the reservedschemaandbranchnames under a file system catalog's table location, and when a schema is validated.docs/source/api/format_table.rstdocumentsFormatTableand points at the generic entry pointsfor everything else.
docs/source/api.rstanddocs/source/user_guide.rstregister the two pagesin their toctrees.
Generative AI tooling
Generated-by: Claude Code (Claude Opus 5)
🤖 Generated with Claude Code