SteNicholas commented on code in PR #222:
URL: https://github.com/apache/paimon-cpp/pull/222#discussion_r3840608348
##########
include/paimon/reader/batch_reader.h:
##########
@@ -47,9 +47,10 @@ class PAIMON_EXPORT BatchReader {
/// avoid potential issues during conversion through the Arrow C Data
Interface.
///
/// @return A result containing a `::ReadBatch`, which consists of a
unique pointer to
- /// `ArrowArray` and a unique pointer to `ArrowSchema`. Returned array
contains a `_VALUE_KIND`
- /// field (the first field) to indicate the row kind of each row. Deleted
or index-filtered rows
- /// are removed.
+ /// `ArrowArray` and a unique pointer to `ArrowSchema`. A reader over a
table that records row
+ /// kinds puts a `_VALUE_KIND` field first to carry them; one over a table
where every row is
+ /// an insert, such as a format table, has no such field. Deleted or
index-filtered rows are
+ /// removed.
virtual Result<ReadBatch> NextBatch() = 0;
Review Comment:
Reverted the contract change. `BatchReader::NextBatch()` promises the
leading `_VALUE_KIND` field again for every table, and `FormatTableRead` wraps
its reader in `CompleteRowKindBatchReader`, so a format table's batches carry
the field filled with inserts and field indexes are unchanged.
`TestReadsCarryTheValueKindField` covers both `NextBatch()` and
`NextBatchWithBitmap()`.
##########
include/paimon/table/format/format_data_split.h:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <limits>
+#include <map>
+#include <string>
+#include <vector>
+
+#include "paimon/table/source/split.h"
+#include "paimon/visibility.h"
+
+namespace paimon {
+
+/// A split of a format table: the data files of one partition directory, or
of the table
+/// directory itself when the table is not partitioned.
+///
+/// Every file is held whole, since parquet and orc each record where their
own row groups and
+/// stripes begin and a reader handed a byte range of one would have to find
that out for itself.
+/// The partition is carried on the split rather than read from the files: a
Hive-style layout
+/// keeps partition values in the directory names.
+class PAIMON_EXPORT FormatDataSplit : public Split {
Review Comment:
The generic entry points now dispatch to a format table:
`TableScan::Create()`, `TableRead::Create()`, `FileStoreWrite::Create()` and
`FileStoreCommit::Create()` recognise one from the schema under the table path,
so a caller holding a path never has to name these types. With that in place
only `FormatTable` itself has to stay public, for `Catalog::GetFormatTable()`.
The one thing that would be lost by moving the rest under `src` is
`FormatTableScan::ListPartitions()`, which has no equivalent on the `TableScan`
interface. So either:
(a) keep them public, as Java keeps `FormatTableScan` / `FormatTableRead` /
`FormatTableWrite` / `FormatTableCommit` public in
`org.apache.paimon.table.format`; or
(b) lift `ListPartitions()` onto `TableScan` and move `FormatTableScan`,
`FormatTableRead`, `FormatTableWrite`, `FormatTableCommit`, `FormatDataSplit`
and `FormatCommitMessage` under `src`.
I am happy to do (b), in this PR or a follow-up. Which would you prefer?
##########
include/paimon/table/format/format_data_split.h:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <limits>
+#include <map>
+#include <string>
+#include <vector>
+
+#include "paimon/table/source/split.h"
+#include "paimon/visibility.h"
+
+namespace paimon {
+
+/// A split of a format table: the data files of one partition directory, or
of the table
+/// directory itself when the table is not partitioned.
+///
+/// Every file is held whole, since parquet and orc each record where their
own row groups and
+/// stripes begin and a reader handed a byte range of one would have to find
that out for itself.
+/// The partition is carried on the split rather than read from the files: a
Hive-style layout
+/// keeps partition values in the directory names.
+class PAIMON_EXPORT FormatDataSplit : public Split {
+ public:
+ /// Frame of `Split::Serialize` for a format table split. This encoding is
paimon-cpp's own
+ /// and has no cross-runtime counterpart, so these bytes travel between
paimon-cpp processes
+ /// only.
+ static constexpr int64_t MAGIC = -8172530964192837451L;
+ static constexpr int32_t VERSION = 1;
+
+ /// One data file of the split.
+ struct PAIMON_EXPORT FileMeta {
+ FileMeta(const std::string& _file_path, int64_t _file_size)
+ : file_path(_file_path), file_size(_file_size) {}
+
+ bool operator==(const FileMeta& other) const {
+ return file_path == other.file_path && file_size ==
other.file_size;
+ }
+
+ /// Bytes this entry accounts for.
+ int64_t ReadSize() const {
+ return file_size;
+ }
Review Comment:
Removed. `FormatDataSplit` and `FormatDataSplit::FileMeta` expose plain
members now (`files`, `partition`, `file_path`, `file_size`); only
`TotalSize()` is left, since it computes rather than returns a member.
##########
include/paimon/table/format/format_table_read.h:
##########
@@ -0,0 +1,109 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <memory>
+#include <optional>
+#include <string>
+#include <vector>
+
+#include "paimon/memory/memory_pool.h"
+#include "paimon/reader/batch_reader.h"
+#include "paimon/result.h"
+#include "paimon/table/format/format_table.h"
+#include "paimon/table/source/split.h"
+#include "paimon/table/source/table_read.h"
+#include "paimon/visibility.h"
+
+namespace paimon {
+
+class Predicate;
+
+/// Reads the splits a `FormatTableScan` produced.
+///
+/// The batches hold the table's own columns only, with no leading
`_VALUE_KIND` field: a directory
+/// of plain data files records no row kind and every row in it is an insert.
Partition columns are
+/// rebuilt from the split's partition values, since the data files do not
carry them.
+///
+/// A batch borrows memory from the reader that produced it, so every batch
must be released before
+/// that reader is destroyed.
+///
+/// Building a reader leaves the read as it was, so one may be shared between
threads; the
+/// `BatchReader`s it hands out may not be, as `TableRead` says.
+///
+/// `CreateCountReader()` is not implemented and falls through to
`TableRead`'s default, which
+/// refuses: counting a format table's rows means reading them.
+class PAIMON_EXPORT FormatTableRead : public TableRead {
+ public:
+ /// @param table Table the splits belong to.
+ /// @param projection Names of the columns to read, in the order they
should appear. When
+ /// absent, every column of the table is read. A column named twice
is rejected.
+ /// @param limit Upper bound on the rows to return across all splits of
one reader. When
+ /// absent, every row is returned.
+ /// @param pool Memory pool the batches are allocated from.
+ /// @param predicate Rows the caller is interested in, as a filter over
the table's columns.
+ /// It is pushed into the file readers so the format skips what its
own statistics let
+ /// it skip; on its own that is a best effort and rows the
predicate rejects can still
+ /// come back. Every field it names must be a column of the table,
of the type the
+ /// conjunct declares, with literals of that same type and none of
them null. The field
+ /// index a conjunct carries is not read - a field is resolved by
name - so a predicate
+ /// built against the table stays valid under any projection.
+ /// @param enable_predicate_filter Whether the returned reader applies
`predicate` exactly to
+ /// the rows it returns. Off by default, as everywhere else in
paimon-cpp. With it on,
+ /// `predicate` may only name columns the projection keeps, since a
column the reader
+ /// does not produce cannot be tested.
+ static Result<std::unique_ptr<FormatTableRead>> Create(
+ const std::shared_ptr<FormatTable>& table,
+ const std::optional<std::vector<std::string>>& projection =
std::nullopt,
+ const std::optional<int32_t>& limit = std::nullopt,
+ const std::shared_ptr<MemoryPool>& pool = nullptr,
+ const std::shared_ptr<Predicate>& predicate = nullptr,
+ bool enable_predicate_filter = false);
Review Comment:
Noted. The defaults are still there only because the header is still public;
if we go with moving these under `src` (see the thread on
`format_data_split.h`) I will drop them in the same change.
Elsewhere in production code the defaults are gone:
`FormatFileNaming::Create()` takes the data file prefix, and
`SchemaValidation::ValidateFormatTableSchema()` takes the file system.
##########
include/paimon/table/source/split.h:
##########
@@ -34,9 +34,12 @@ namespace paimon {
class MemoryPool;
/// An input split for reading operation. Needed by most batch computation
engines. Support
-/// Serialize and Deserialize, compatible with java version.
-/// This split can be either a `DataSplit` (for direct data file reads) or an
`IndexedSplit`
-/// (for reads leveraging global indexes).
+/// Serialize and Deserialize.
+///
+/// This split can be a `DataSplit` (for direct data file reads), an
`IndexedSplit` (for reads
+/// leveraging global indexes), or a `FormatDataSplit` (for a format table's
plain data files).
+/// The first two use the cross-runtime encoding; `FormatDataSplit` has none,
so it round-trips
+/// only here.
Review Comment:
Agreed, and removed. `FormatDataSplit` and `FormatCommitMessage` have no
serialized form now: `Split::Serialize()` and `CommitMessage::Serialize()`
return `NotImplemented` saying a format table's plan is in-memory only, so a
plan and a commit message are made and used within one process.
`TestASplitIsNotSerializable` and `TestACommitMessageIsNotSerializable` cover
both.
##########
src/paimon/common/reader/predicate_batch_reader.cpp:
##########
@@ -81,8 +81,16 @@ Result<BatchReader::ReadBatchWithBitmap>
PredicateBatchReader::NextBatchWithBitm
auto& [batch, bitmap] = batch_with_bitmap;
auto& [c_array, c_schema] = batch;
assert(c_array);
- PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Array> array,
- arrow::ImportArray(c_array.get(),
c_schema.get()));
+ // Importing takes the batch over on success only. A
`unique_ptr<ArrowArray>` frees the
+ // struct without releasing what it points at, so a failure here has
to release the batch
+ // itself or the reader below is left holding memory nothing will ever
hand back.
+ arrow::Result<std::shared_ptr<arrow::Array>> imported =
+ arrow::ImportArray(c_array.get(), c_schema.get());
+ if (!imported.ok()) {
+ ReaderUtils::ReleaseReadBatch(std::move(batch));
+ return ToPaimonStatus(imported.status());
+ }
+ std::shared_ptr<arrow::Array> array =
std::move(imported).ValueUnsafe();
Review Comment:
You are right, it was not a problem found in testing. Reverted to
`PAIMON_ASSIGN_OR_RAISE_FROM_ARROW`.
##########
src/paimon/common/utils/serialization_utils.h:
##########
@@ -47,6 +49,50 @@ class SerializationUtils {
SerializationUtils() = delete;
~SerializationUtils() = delete;
+ /// Longest string this encoding can carry:
`MemorySegmentOutputStream::WriteString` prefixes a
+ /// string with a 16-bit length and `DataInputStream::ReadString` reads it
back unsigned, so a
+ /// longer one would be written in full behind a truncated length and read
back as a different
+ /// string.
+ static constexpr size_t MAX_SERIALIZED_STRING_BYTES = 65535;
+
+ /// Fails when `value` could not be written and read back as itself. Meant
to run before a
+ /// byte is written, so that the failure happens where the bytes came from.
+ ///
+ /// @param what What the string is, for the error message.
+ /// @param container What it is part of, so the message says which
encoding refused it.
+ static Status CheckSerializableString(const std::string& value, const
char* what,
+ const char* container) {
+ if (value.size() > MAX_SERIALIZED_STRING_BYTES) {
+ return Status::Invalid(
+ fmt::format("{} is {} bytes, more than the {} a serialized {}
can carry", what,
+ value.size(), MAX_SERIALIZED_STRING_BYTES,
container));
+ }
+ return Status::OK();
+ }
+
+ /// Rejects an element count that the bytes left to read could not
possibly hold.
+ ///
+ /// The bytes are untrusted, so a negative count, or one the rest of the
stream could not
+ /// encode, must come back as a `Status` rather than reach a `reserve()`
that would throw past
+ /// the error model. `min_bytes_per_element` is what makes the check
tight: an entry costs
+ /// several bytes, so comparing against the bytes left one-for-one would
let a count many
+ /// times larger than the input through.
+ static Status CheckElementCount(int32_t count, int64_t
min_bytes_per_element, const char* what,
+ DataInputStream* in) {
+ if (count < 0) {
Review Comment:
Moot now: those helpers went away with the split serialization.
##########
src/paimon/core/table/table.cpp:
##########
@@ -44,6 +48,28 @@ Result<std::shared_ptr<Table>> Table::Create(const
std::shared_ptr<FileSystem>&
fmt::format("load table schema for {} failed",
identifier.ToString()));
}
+ // Only a managed table is a `Table`; describing another type as one would
hand back snapshots
+ // and manifests it never had. Checked here because this is the one place
every `Table` built
+ // from a table directory passes through, catalogs included.
+ PAIMON_ASSIGN_OR_RAISE(TableType table_type,
+
TableTypeDefine::FromOptions((*latest_schema)->Options()));
+ if (table_type == TableType::FORMAT_TABLE) {
+ return Status::Invalid(
+ fmt::format("Cannot open format table '{}' as a Table, please use "
+ "'Catalog::GetFormatTable' or 'FormatTable::Create'.",
+ identifier.ToString()));
+ }
+ // A materialized table is a managed table that also carries the SQL it
materializes.
+ if (table_type != TableType::TABLE && table_type !=
TableType::MATERIALIZED_TABLE) {
+ const std::map<std::string, std::string>& options =
(*latest_schema)->Options();
+ auto type_iter = options.find(Options::TYPE);
+ return Status::NotImplemented(fmt::format(
+ "Cannot open table '{}': its '{}' is '{}', a table type paimon-cpp
does not "
+ "implement, and opening it as a managed table would look for
snapshots it never had.",
+ identifier.ToString(), Options::TYPE,
+ type_iter == options.end() ? std::string() : type_iter->second));
+ }
Review Comment:
Done. `Table::Create()` calls `CatalogUtils::CheckManagedTableType()`, so
the wording and the status codes come from one place.
`TableTest.TestCreateRejectsATableTypeThatIsNotAManagedTable` covers it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
