Google Git Google Git
Sign in
apache/paimon-cpp/refs/heads/release-0.4/./test/inte/blob_table_inte_test.cpp
blob: fd706e640c7a5d47b16076176f7455a9a580632f [file]
/*
* 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.
*/
#include <algorithm>
#include <cstdint>
#include <cstdlib>
#include <initializer_list>
#include <map>
#include <memory>
#include <numeric>
#include <optional>
#include <set>
#include <string>
#include <string_view>
#include <utility>
#include <vector>
#include "arrow/api.h"
#include "arrow/array/array_base.h"
#include "arrow/array/builder_binary.h"
#include "arrow/array/builder_nested.h"
#include "arrow/array/builder_primitive.h"
#include "arrow/c/abi.h"
#include "arrow/c/bridge.h"
#include "arrow/ipc/json_simple.h"
#include "arrow/type.h"
#include "gtest/gtest.h"
#include "paimon/commit_context.h"
#include "paimon/common/data/binary_array_writer.h"
#include "paimon/common/data/binary_row.h"
#include "paimon/common/data/binary_row_writer.h"
#include "paimon/common/data/blob_defs.h"
#include "paimon/common/data/blob_utils.h"
#include "paimon/common/data/blob_view_struct.h"
#include "paimon/common/factories/io_hook.h"
#include "paimon/common/reader/reader_utils.h"
#include "paimon/common/table/special_fields.h"
#include "paimon/common/utils/checked_cast.h"
#include "paimon/common/utils/path_util.h"
#include "paimon/common/utils/scope_guard.h"
#include "paimon/core/global_index/indexed_split_impl.h"
#include "paimon/core/snapshot.h"
#include "paimon/core/stats/simple_stats.h"
#include "paimon/core/table/source/data_split_impl.h"
#include "paimon/core/table/source/plan_impl.h"
#include "paimon/core/utils/file_utils.h"
#include "paimon/core/utils/snapshot_manager.h"
#include "paimon/data/blob.h"
#include "paimon/data/blob_descriptor.h"
#include "paimon/defs.h"
#include "paimon/file_store_write.h"
#include "paimon/format/mosaic/mosaic_format_defs.h"
#include "paimon/fs/file_system.h"
#include "paimon/fs/local/local_file_system.h"
#include "paimon/global_index/bitmap_global_index_result.h"
#include "paimon/memory/memory_pool.h"
#include "paimon/predicate/literal.h"
#include "paimon/predicate/predicate_builder.h"
#include "paimon/read_context.h"
#include "paimon/reader/batch_reader.h"
#include "paimon/record_batch.h"
#include "paimon/result.h"
#include "paimon/scan_context.h"
#include "paimon/status.h"
#include "paimon/table/source/data_split.h"
#include "paimon/table/source/startup_mode.h"
#include "paimon/table/source/table_read.h"
#include "paimon/testing/utils/deletion_vector_test_helper.h"
#include "paimon/testing/utils/io_exception_helper.h"
#include "paimon/testing/utils/test_helper.h"
#include "paimon/testing/utils/testharness.h"
#include "paimon/write_context.h"
namespace paimon {
class DataSplit;
class RecordBatch;
} // namespace paimon
namespace paimon::test {
class BlobTableInteTest : public testing::Test, public ::testing::WithParamInterface<std::string> {
public:
void SetUp() override {
pool_ = GetDefaultPool();
dir_ = UniqueTestDirectory::Create("local");
blob_dir_ = UniqueTestDirectory::Create("local");
}
void TearDown() override {
dir_.reset();
}
void CreateTable(const std::vector<std::string>& partition_keys,
const std::map<std::string, std::string>& options) const {
CreateTable(fields_, partition_keys, options);
}
void CreateTable(const arrow::FieldVector& fields,
const std::vector<std::string>& partition_keys,
const std::map<std::string, std::string>& options) const {
auto schema = arrow::schema(fields);
::ArrowSchema c_schema;
ASSERT_TRUE(arrow::ExportSchema(*schema, &c_schema).ok());
ASSERT_OK_AND_ASSIGN(auto catalog, Catalog::Create(dir_->Str(), {}));
ASSERT_OK(catalog->CreateDatabase("foo", {}, /*ignore_if_exists=*/false));
ASSERT_OK(catalog->CreateTable(Identifier("foo", "bar"), &c_schema, partition_keys,
/*primary_keys=*/{}, options,
/*ignore_if_exists=*/false));
}
void CreateTable(const std::vector<std::string>& partition_keys) const {
std::map<std::string, std::string> options = {{Options::FILE_FORMAT, GetParam()},
{Options::FILE_SYSTEM, "local"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"}};
return CreateTable(partition_keys, options);
}
void CreateTable() const {
return CreateTable(/*partition_keys=*/{});
}
Result<std::vector<std::shared_ptr<CommitMessage>>> WriteArray(
const std::string& table_path, const std::map<std::string, std::string>& partition,
const std::vector<std::string>& write_cols,
const std::vector<std::shared_ptr<arrow::Array>>& write_arrays) const {
// write
WriteContextBuilder write_builder(table_path, "commit_user_1");
write_builder.WithWriteSchema(write_cols);
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<WriteContext> write_context, write_builder.Finish());
PAIMON_ASSIGN_OR_RAISE(auto file_store_write,
FileStoreWrite::Create(std::move(write_context)));
for (const auto& write_array : write_arrays) {
ArrowArray c_array;
PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*write_array, &c_array));
auto record_batch = std::make_unique<RecordBatch>(
partition, /*bucket=*/0,
/*row_kinds=*/std::vector<RecordBatch::RowKind>(), &c_array);
PAIMON_RETURN_NOT_OK(file_store_write->Write(std::move(record_batch)));
}
PAIMON_ASSIGN_OR_RAISE(auto commit_msgs,
file_store_write->PrepareCommit(
/*wait_compaction=*/false, /*commit_identifier=*/0));
PAIMON_RETURN_NOT_OK(file_store_write->Close());
return commit_msgs;
}
void SetFirstRowId(int64_t reset_first_row_id,
std::vector<std::shared_ptr<CommitMessage>>& commit_msgs) const {
for (auto& commit_msg : commit_msgs) {
auto commit_msg_impl = std::dynamic_pointer_cast<CommitMessageImpl>(commit_msg);
ASSERT_TRUE(commit_msg_impl);
for (auto& file : commit_msg_impl->data_increment_.new_files_) {
file->AssignFirstRowId(reset_first_row_id);
}
}
}
Status Commit(const std::string& table_path,
const std::vector<std::shared_ptr<CommitMessage>>& commit_msgs) const {
// commit
CommitContextBuilder commit_builder(table_path, "commit_user_1");
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<CommitContext> commit_context,
commit_builder.Finish());
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileStoreCommit> file_store_commit,
FileStoreCommit::Create(std::move(commit_context)));
return file_store_commit->Commit(commit_msgs);
}
/// DeletionVectorTestHelper::CreateDeletionVectorCommitMessage, documented there, plus the
/// commit of the message it returns.
Result<std::shared_ptr<CommitMessage>> CommitDeletionVectors(
const std::string& table_path, const std::shared_ptr<CommitMessage>& base_commit_msg,
const std::map<std::string, std::vector<int64_t>>& deleted_positions_by_anchor,
const std::shared_ptr<CommitMessage>& replaced_commit_msg = nullptr) const {
PAIMON_ASSIGN_OR_RAISE(
std::shared_ptr<CommitMessage> commit_message,
DeletionVectorTestHelper::CreateDeletionVectorCommitMessage(
dir_->GetFileSystem(), table_path, /*file_format_identifier=*/GetParam(),
base_commit_msg, deleted_positions_by_anchor, pool_, replaced_commit_msg));
PAIMON_RETURN_NOT_OK(Commit(table_path, {commit_message}));
return commit_message;
}
/// Creates the (f0 INT, view) blob-view table the dangling-reference tests share: unaware
/// bucket and data evolution on. Returns the table's fields, which the callers reuse to
/// build their expected arrays.
arrow::FieldVector CreateBlobViewTable(bool deletion_vectors_enabled) const {
arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
BlobUtils::ToArrowField("view", true)};
std::map<std::string, std::string> options = {
{Options::FILE_FORMAT, GetParam()}, {Options::BUCKET, "-1"},
{Options::ROW_TRACKING_ENABLED, "true"}, {Options::DATA_EVOLUTION_ENABLED, "true"},
{Options::BLOB_VIEW_FIELD, "view"}, {Options::FILE_SYSTEM, "local"}};
if (deletion_vectors_enabled) {
options.emplace(Options::DELETION_VECTORS_ENABLED, "true");
}
CreateTable(fields, /*partition_keys=*/{}, options);
return fields;
}
/// Writes and commits the two rows the blob-view skip tests share: row 0 holds a blob view
/// reference to an upstream table that does not exist, row 1 holds no blob view at all.
/// Resolving row 0 fails the whole read, so any read the pre-read still reaches it from must
/// error; only dropping the row through the deletion vector or the row ranges avoids that,
/// because those are the two filters the pre-read applies. The table must already exist with
/// fields (f0, view).
///
/// `first_row_id`, when set, is stamped on the returned metas before committing: the commit
/// assigns row ids onto its own copies, so a caller that needs them (to look up the group's
/// anchor) has to set them here, matching what the commit would assign.
Result<std::vector<std::shared_ptr<CommitMessage>>> WriteDanglingBlobViewRows(
const std::string& table_path,
const std::optional<int64_t>& first_row_id = std::nullopt) const {
Identifier upstream_identifier("nonexistent_db", "nonexistent_table");
BlobViewStruct view_struct(upstream_identifier, /*field_id=*/2, /*row_id=*/0);
PAIMON_ASSIGN_OR_RAISE(PAIMON_UNIQUE_PTR<Bytes> serialized, view_struct.Serialize(pool_));
arrow::LargeBinaryBuilder view_builder;
PAIMON_RETURN_NOT_OK_FROM_ARROW(view_builder.Append(
reinterpret_cast<const uint8_t*>(serialized->data()), serialized->size()));
PAIMON_RETURN_NOT_OK_FROM_ARROW(view_builder.AppendNull());
std::shared_ptr<arrow::Array> write_view_array;
PAIMON_RETURN_NOT_OK_FROM_ARROW(view_builder.Finish(&write_view_array));
PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
std::shared_ptr<arrow::Array> write_f0_array,
arrow::ipc::internal::json::ArrayFromJSON(arrow::int32(), R"([100, 101])"));
PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
std::shared_ptr<arrow::Array> write_struct,
arrow::StructArray::Make(arrow::ArrayVector({write_f0_array, write_view_array}),
std::vector<std::string>({"f0", "view"})));
PAIMON_ASSIGN_OR_RAISE(
std::vector<std::shared_ptr<CommitMessage>> commit_msgs,
WriteArray(table_path, /*partition=*/{}, {"f0", "view"}, {write_struct}));
if (first_row_id) {
SetFirstRowId(first_row_id.value(), commit_msgs);
}
PAIMON_RETURN_NOT_OK(Commit(table_path, commit_msgs));
return commit_msgs;
}
Status WriteNextSchema(const std::string& table_path, const std::vector<DataField>& fields,
int32_t highest_field_id,
const std::map<std::string, std::string>& options) const {
return TestHelper::WriteNextSchema(dir_->GetFileSystem(), table_path, fields,
highest_field_id, options);
}
/// Scan table and return the plan (without reading data).
Result<std::shared_ptr<Plan>> ScanTable(const std::string& table_path,
const std::shared_ptr<Predicate>& predicate = nullptr,
const std::vector<Range>& row_ranges = {}) const {
ScanContextBuilder scan_context_builder(table_path);
scan_context_builder.SetPredicate(predicate);
if (!row_ranges.empty()) {
auto global_index_result = BitmapGlobalIndexResult::FromRanges(row_ranges);
scan_context_builder.SetGlobalIndexResult(global_index_result);
}
PAIMON_ASSIGN_OR_RAISE(auto scan_context, scan_context_builder.Finish());
PAIMON_ASSIGN_OR_RAISE(auto table_scan, TableScan::Create(std::move(scan_context)));
PAIMON_ASSIGN_OR_RAISE(auto result_plan, table_scan->CreatePlan());
return result_plan;
}
/// Read from table using a pre-scanned plan and return data after the reader is destroyed.
Result<std::shared_ptr<arrow::ChunkedArray>> ReadTable(
const std::string& table_path, const std::vector<std::string>& read_schema,
const std::shared_ptr<Plan>& plan, const std::shared_ptr<Predicate>& predicate = nullptr,
const std::map<std::string, std::string>& options = {}) const {
auto splits = plan->Splits();
ReadContextBuilder read_context_builder(table_path);
read_context_builder.SetReadFieldNames(read_schema).SetPredicate(predicate);
read_context_builder.EnableLateMaterializing(false);
if (!options.empty()) {
read_context_builder.SetOptions(options);
}
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ReadContext> read_context,
read_context_builder.Finish());
PAIMON_ASSIGN_OR_RAISE(auto table_read, TableRead::Create(std::move(read_context)));
PAIMON_ASSIGN_OR_RAISE(auto batch_reader, table_read->CreateReader(splits));
PAIMON_ASSIGN_OR_RAISE(auto read_result,
ReadResultCollector::CollectResult(std::move(batch_reader)));
return read_result;
}
/// Convenience: scan + read in one call.
Result<std::shared_ptr<arrow::ChunkedArray>> ScanAndReadResult(
const std::string& table_path, const std::vector<std::string>& read_schema,
const std::shared_ptr<Predicate>& predicate = nullptr,
const std::vector<Range>& row_ranges = {}) const {
PAIMON_ASSIGN_OR_RAISE(auto result_plan, ScanTable(table_path, predicate, row_ranges));
return ReadTable(table_path, read_schema, result_plan, predicate);
}
/// Prepend a _VALUE_KIND (Insert) column to a StructArray.
static Result<std::shared_ptr<arrow::StructArray>> PrependRowKindColumn(
const std::shared_ptr<arrow::StructArray>& array) {
auto row_kind_scalar =
std::make_shared<arrow::Int8Scalar>(RowKind::Insert()->ToByteValue());
PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
auto row_kind_array, arrow::MakeArrayFromScalar(*row_kind_scalar, array->length()));
arrow::ArrayVector fields_with_row_kind = array->fields();
std::vector<std::string> names_with_row_kind =
arrow::schema(array->type()->fields())->field_names();
fields_with_row_kind.insert(fields_with_row_kind.begin(), row_kind_array);
names_with_row_kind.insert(names_with_row_kind.begin(), "_VALUE_KIND");
PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
auto result, arrow::StructArray::Make(fields_with_row_kind, names_with_row_kind));
return std::dynamic_pointer_cast<arrow::StructArray>(result);
}
Status ScanAndRead(const std::string& table_path, const std::vector<std::string>& read_schema,
const std::shared_ptr<arrow::StructArray>& expected_array,
const std::shared_ptr<Predicate>& predicate = nullptr,
const std::vector<Range>& row_ranges = {}) const {
PAIMON_ASSIGN_OR_RAISE(auto scan_read,
ScanAndReadResult(table_path, read_schema, predicate, row_ranges));
if (!expected_array) {
EXPECT_FALSE(scan_read);
return Status::OK();
}
PAIMON_ASSIGN_OR_RAISE(auto expected_with_row_kind, PrependRowKindColumn(expected_array));
auto expected_chunk_array = std::make_shared<arrow::ChunkedArray>(expected_with_row_kind);
EXPECT_TRUE(expected_chunk_array->Equals(scan_read))
<< "result:" << scan_read->ToString() << std::endl
<< "expected:" << expected_chunk_array->ToString();
return Status::OK();
}
std::shared_ptr<arrow::StructArray> PrepareBulkData(
int32_t write_batch_size, std::function<std::string(int32_t)> data_generator,
const arrow::FieldVector& fields) const {
std::string data_str = "[";
for (int32_t i = 0; i < write_batch_size; i++) {
data_str.append("[");
auto row_str = data_generator(i);
data_str.append(row_str);
data_str.append("],");
}
data_str.pop_back();
data_str.append("]");
return std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), data_str)
.ValueOrDie());
}
/// Convert a StructArray with raw blob bytes into a StructArray with serialized
/// BlobDescriptor bytes. Each raw blob value is written to a temporary file, and
/// the corresponding cell is replaced with the serialized BlobDescriptor pointing
/// to that file.
/// Common framework for transforming blob fields in a StructArray.
/// Non-blob fields are kept as-is; blob fields are processed row-by-row via `transform_row`.
/// `transform_row` receives (binary_value_view) and returns the transformed bytes via builder.
using BlobRowTransform =
std::function<Status(const std::string_view& value, arrow::LargeBinaryBuilder* builder)>;
Result<std::shared_ptr<arrow::StructArray>> TransformBlobFields(
const std::shared_ptr<arrow::StructArray>& input_array,
const std::set<std::string>& blob_fields, BlobRowTransform transform_row) const {
auto fields = input_array->type()->fields();
arrow::ArrayVector child_arrays;
for (const auto& field : fields) {
auto col = input_array->GetFieldByName(field->name());
if (blob_fields.count(field->name()) == 0) {
child_arrays.push_back(col);
continue;
}
const auto& binary_array = checked_cast<const arrow::LargeBinaryArray&>(*col);
arrow::LargeBinaryBuilder builder;
for (int64_t i = 0; i < binary_array.length(); ++i) {
if (binary_array.IsNull(i)) {
PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.AppendNull());
continue;
}
PAIMON_RETURN_NOT_OK(transform_row(binary_array.GetView(i), &builder));
}
std::shared_ptr<arrow::Array> result_col;
PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Finish(&result_col));
child_arrays.push_back(result_col);
}
PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(auto result,
arrow::StructArray::Make(child_arrays, fields));
return result;
}
Result<std::shared_ptr<arrow::StructArray>> ConvertRawBlobToDescriptor(
const std::shared_ptr<arrow::StructArray>& raw_array,
const std::set<std::string>& blob_fields) {
auto fs = std::make_shared<LocalFileSystem>();
return TransformBlobFields(
raw_array, blob_fields,
[&](const std::string_view& raw_value, arrow::LargeBinaryBuilder* builder) -> Status {
std::string file_path =
blob_dir_->Str() + "/blob_" + std::to_string(blob_file_counter_++) + ".bin";
auto raw_size = static_cast<int64_t>(raw_value.size());
PAIMON_ASSIGN_OR_RAISE(auto out, fs->Create(file_path, /*overwrite=*/true));
PAIMON_ASSIGN_OR_RAISE(auto written, out->Write(raw_value.data(), raw_size));
PAIMON_RETURN_NOT_OK(out->Flush());
PAIMON_RETURN_NOT_OK(out->Close());
if (written != raw_size) {
return Status::Invalid("Short write: expected {}, wrote {}", raw_value.size(),
written);
}
PAIMON_ASSIGN_OR_RAISE(auto blob, Blob::FromPath(file_path));
auto descriptor = blob->ToDescriptor(pool_);
PAIMON_RETURN_NOT_OK_FROM_ARROW(
builder->Append(descriptor->data(), descriptor->size()));
return Status::OK();
});
}
/// Delete the file referenced by the serialized BlobDescriptor at `row` of blob field
/// `field_name`, so that a subsequent table write observes a missing file.
Status DeleteDescriptorTarget(const std::shared_ptr<arrow::StructArray>& desc_array,
const std::string& field_name, int64_t row) const {
const auto& blob_col =
checked_cast<const arrow::LargeBinaryArray&>(*desc_array->GetFieldByName(field_name));
std::string_view descriptor_bytes = blob_col.GetView(row);
PAIMON_ASSIGN_OR_RAISE(
std::unique_ptr<BlobDescriptor> descriptor,
BlobDescriptor::Deserialize(descriptor_bytes.data(), descriptor_bytes.size()));
LocalFileSystem fs;
return fs.Delete(descriptor->Uri(), /*recursive=*/false);
}
/// Rewrite the serialized BlobDescriptor at `row` of blob field `field_name` to reference
/// an offset beyond the end of the target file, so that a subsequent table write observes
/// a fetch failure that is not a missing file. `row` counts non-null rows.
Result<std::shared_ptr<arrow::StructArray>> CorruptDescriptorOffset(
const std::shared_ptr<arrow::StructArray>& desc_array, const std::string& field_name,
int64_t row) const {
int64_t current_row = 0;
return TransformBlobFields(
desc_array, {field_name},
[&](const std::string_view& descriptor_bytes,
arrow::LargeBinaryBuilder* builder) -> Status {
if (current_row++ != row) {
PAIMON_RETURN_NOT_OK_FROM_ARROW(
builder->Append(descriptor_bytes.data(), descriptor_bytes.size()));
return Status::OK();
}
PAIMON_ASSIGN_OR_RAISE(
std::unique_ptr<BlobDescriptor> descriptor,
BlobDescriptor::Deserialize(descriptor_bytes.data(), descriptor_bytes.size()));
PAIMON_ASSIGN_OR_RAISE(
std::unique_ptr<BlobDescriptor> corrupted,
BlobDescriptor::Create(descriptor->Version(), descriptor->Uri(),
/*offset=*/1 << 20, descriptor->Length()));
auto corrupted_bytes = corrupted->Serialize(pool_);
PAIMON_RETURN_NOT_OK_FROM_ARROW(
builder->Append(corrupted_bytes->data(), corrupted_bytes->size()));
return Status::OK();
});
}
struct BlobDescriptorPathRewrite {
std::string table_path;
std::vector<std::string> table_relative_blob_dirs;
};
static std::optional<std::string> TryRewriteDescriptorUri(
const std::string& descriptor_uri, const BlobDescriptorPathRewrite& rewrite,
const std::shared_ptr<LocalFileSystem>& fs) {
if (rewrite.table_path.empty()) {
return std::nullopt;
}
for (const auto& blob_dir : rewrite.table_relative_blob_dirs) {
const std::string marker = "/" + blob_dir + "/";
auto marker_pos = descriptor_uri.find(marker);
if (marker_pos != std::string::npos) {
std::string relative_blob_path = descriptor_uri.substr(marker_pos + 1);
return PathUtil::JoinPath(rewrite.table_path, relative_blob_path);
}
}
return std::nullopt;
}
/// Convert a StructArray with serialized BlobDescriptor bytes back to a StructArray
/// with raw blob bytes. Only blob fields are resolved; other columns (including
/// _VALUE_KIND) are kept as-is.
Result<std::shared_ptr<arrow::StructArray>> ConvertDescriptorToRawBlob(
const std::shared_ptr<arrow::StructArray>& desc_array,
const std::set<std::string>& blob_fields,
const BlobDescriptorPathRewrite& rewrite = {}) const {
auto fs = std::make_shared<LocalFileSystem>();
return TransformBlobFields(
desc_array, blob_fields,
[&](const std::string_view& descriptor_bytes,
arrow::LargeBinaryBuilder* builder) -> Status {
PAIMON_ASSIGN_OR_RAISE(
auto descriptor,
BlobDescriptor::Deserialize(descriptor_bytes.data(), descriptor_bytes.size()));
std::string descriptor_uri = descriptor->Uri();
auto rewritten_uri = TryRewriteDescriptorUri(descriptor_uri, rewrite, fs);
if (rewritten_uri.has_value()) {
descriptor_uri = rewritten_uri.value();
}
PAIMON_ASSIGN_OR_RAISE(
auto rewritten_descriptor,
BlobDescriptor::Create(descriptor->Version(), descriptor_uri,
descriptor->Offset(), descriptor->Length()));
auto rewritten_descriptor_bytes = rewritten_descriptor->Serialize(pool_);
PAIMON_ASSIGN_OR_RAISE(auto blob,
Blob::FromDescriptor(rewritten_descriptor_bytes->data(),
rewritten_descriptor_bytes->size()));
PAIMON_ASSIGN_OR_RAISE(auto data, blob->ToData(fs, pool_));
PAIMON_RETURN_NOT_OK_FROM_ARROW(builder->Append(data->data(), data->size()));
return Status::OK();
});
}
Result<std::shared_ptr<arrow::Array>> ResolveBlobDescriptors(
const std::shared_ptr<arrow::Array>& values_array) const {
const auto& values = checked_cast<const arrow::LargeBinaryArray&>(*values_array);
auto fs = std::make_shared<LocalFileSystem>();
arrow::LargeBinaryBuilder builder;
for (int64_t i = 0; i < values.length(); ++i) {
if (values.IsNull(i)) {
PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.AppendNull());
continue;
}
std::string_view stored = values.GetView(i);
PAIMON_ASSIGN_OR_RAISE(
std::unique_ptr<Blob> blob,
Blob::FromDescriptor(stored.data(), static_cast<int64_t>(stored.size())));
PAIMON_ASSIGN_OR_RAISE(PAIMON_UNIQUE_PTR<Bytes> data, blob->ToData(fs, pool_));
PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Append(data->data(), data->size()));
}
std::shared_ptr<arrow::Array> normalized_values;
PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Finish(&normalized_values));
return normalized_values;
}
Result<std::shared_ptr<arrow::MapArray>> NormalizeMapBlobValues(
const std::shared_ptr<arrow::MapArray>& map_array, bool blob_as_descriptor) const {
if (!blob_as_descriptor) {
return map_array;
}
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> normalized_values,
ResolveBlobDescriptors(map_array->items()));
return std::make_shared<arrow::MapArray>(map_array->type(), map_array->length(),
map_array->value_offsets(), map_array->keys(),
normalized_values, map_array->null_bitmap(),
map_array->null_count(), map_array->offset());
}
Result<std::shared_ptr<arrow::ListArray>> NormalizeArrayBlobValues(
const std::shared_ptr<arrow::ListArray>& list_array, bool blob_as_descriptor) const {
if (!blob_as_descriptor) {
return list_array;
}
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> normalized_values,
ResolveBlobDescriptors(list_array->values()));
return std::make_shared<arrow::ListArray>(list_array->type(), list_array->length(),
list_array->value_offsets(), normalized_values,
list_array->null_bitmap(),
list_array->null_count(), list_array->offset());
}
using BlobElements = std::vector<std::optional<std::string>>;
static Result<std::shared_ptr<arrow::Array>> MakeArrayBlobArray(
const std::shared_ptr<arrow::DataType>& list_type,
const std::vector<std::optional<BlobElements>>& rows) {
arrow::ListBuilder list_builder(arrow::default_memory_pool(),
std::make_shared<arrow::LargeBinaryBuilder>(), list_type);
auto* element_builder =
checked_cast<arrow::LargeBinaryBuilder*>(list_builder.value_builder());
for (const std::optional<BlobElements>& row : rows) {
if (!row) {
PAIMON_RETURN_NOT_OK_FROM_ARROW(list_builder.AppendNull());
continue;
}
PAIMON_RETURN_NOT_OK_FROM_ARROW(list_builder.Append());
for (const std::optional<std::string>& element : *row) {
if (element) {
PAIMON_RETURN_NOT_OK_FROM_ARROW(element_builder->Append(*element));
} else {
PAIMON_RETURN_NOT_OK_FROM_ARROW(element_builder->AppendNull());
}
}
}
std::shared_ptr<arrow::Array> array;
PAIMON_RETURN_NOT_OK_FROM_ARROW(list_builder.Finish(&array));
return array;
}
using MapBlobEntries = std::vector<std::pair<std::string, std::optional<std::string>>>;
static Result<std::shared_ptr<arrow::Array>> MakeMapBlobArray(
const std::shared_ptr<arrow::DataType>& map_type,
const std::vector<std::optional<MapBlobEntries>>& rows) {
arrow::MapBuilder map_builder(arrow::default_memory_pool(),
std::make_shared<arrow::StringBuilder>(),
std::make_shared<arrow::LargeBinaryBuilder>(), map_type);
auto* key_builder = checked_cast<arrow::StringBuilder*>(map_builder.key_builder());
auto* item_builder = checked_cast<arrow::LargeBinaryBuilder*>(map_builder.item_builder());
for (const std::optional<MapBlobEntries>& row : rows) {
if (!row) {
PAIMON_RETURN_NOT_OK_FROM_ARROW(map_builder.AppendNull());
continue;
}
PAIMON_RETURN_NOT_OK_FROM_ARROW(map_builder.Append());
for (const auto& [key, value] : *row) {
PAIMON_RETURN_NOT_OK_FROM_ARROW(key_builder->Append(key));
if (value) {
PAIMON_RETURN_NOT_OK_FROM_ARROW(item_builder->Append(*value));
} else {
PAIMON_RETURN_NOT_OK_FROM_ARROW(item_builder->AppendNull());
}
}
}
std::shared_ptr<arrow::Array> array;
PAIMON_RETURN_NOT_OK_FROM_ARROW(map_builder.Finish(&array));
return array;
}
void CheckArrayBlobColumn(const std::shared_ptr<arrow::StructArray>& rows,
const std::string& field_name, const std::string& expected_json,
bool blob_as_descriptor) const {
auto list_array =
std::dynamic_pointer_cast<arrow::ListArray>(rows->GetFieldByName(field_name));
ASSERT_TRUE(list_array) << field_name;
ASSERT_OK_AND_ASSIGN(auto normalized,
NormalizeArrayBlobValues(list_array, blob_as_descriptor));
auto expected = arrow::ipc::internal::json::ArrayFromJSON(list_array->type(), expected_json)
.ValueOrDie();
ASSERT_TRUE(expected->Equals(normalized))
<< field_name << " expected: " << expected->ToString()
<< " actual: " << normalized->ToString();
}
void CheckMapBlobColumn(const std::shared_ptr<arrow::StructArray>& rows,
const std::string& field_name, const std::string& expected_json,
bool blob_as_descriptor) const {
auto map_array =
std::dynamic_pointer_cast<arrow::MapArray>(rows->GetFieldByName(field_name));
ASSERT_TRUE(map_array) << field_name;
ASSERT_OK_AND_ASSIGN(auto normalized,
NormalizeMapBlobValues(map_array, blob_as_descriptor));
auto expected = arrow::ipc::internal::json::ArrayFromJSON(map_array->type(), expected_json)
.ValueOrDie();
ASSERT_TRUE(expected->Equals(normalized))
<< field_name << " expected: " << expected->ToString()
<< " actual: " << normalized->ToString();
}
/// Verify DataFileMeta properties from a scan plan.
/// Each vector element corresponds to one expected DataFileMeta (ordered by file index).
static void VerifyDataFileMetas(
const std::shared_ptr<Plan>& plan, size_t expected_file_count,
const std::vector<int64_t>& expected_row_counts,
const std::vector<int64_t>& expected_min_seqs,
const std::vector<int64_t>& expected_max_seqs,
const std::vector<int64_t>& expected_first_row_ids,
const std::vector<std::optional<std::vector<std::string>>>& expected_write_cols) {
std::vector<std::shared_ptr<DataFileMeta>> all_files;
for (const auto& split : plan->Splits()) {
auto data_split = std::dynamic_pointer_cast<DataSplitImpl>(split);
ASSERT_TRUE(data_split);
for (const auto& file : data_split->DataFiles()) {
all_files.push_back(file);
}
}
ASSERT_EQ(all_files.size(), expected_file_count);
ASSERT_EQ(expected_row_counts.size(), expected_file_count);
ASSERT_EQ(expected_min_seqs.size(), expected_file_count);
ASSERT_EQ(expected_max_seqs.size(), expected_file_count);
ASSERT_EQ(expected_first_row_ids.size(), expected_file_count);
ASSERT_EQ(expected_write_cols.size(), expected_file_count);
for (size_t i = 0; i < all_files.size(); ++i) {
const auto& file = all_files[i];
EXPECT_EQ(file->row_count, expected_row_counts[i]);
EXPECT_EQ(file->min_sequence_number, expected_min_seqs[i]);
EXPECT_EQ(file->max_sequence_number, expected_max_seqs[i]);
ASSERT_TRUE(file->first_row_id.has_value());
EXPECT_EQ(file->first_row_id.value(), expected_first_row_ids[i]);
EXPECT_EQ(file->write_cols, expected_write_cols[i]);
}
}
private:
std::shared_ptr<MemoryPool> pool_;
std::unique_ptr<UniqueTestDirectory> dir_;
std::unique_ptr<UniqueTestDirectory> blob_dir_;
int blob_file_counter_ = 0;
arrow::FieldVector fields_ = {arrow::field("f0", arrow::int32()), BlobUtils::ToArrowField("f1"),
arrow::field("f2", arrow::utf8())};
};
std::vector<std::string> GetTestValuesForBlobTableInteTest() {
std::vector<std::string> values;
values.emplace_back("parquet");
#ifdef PAIMON_ENABLE_MOSAIC
values.emplace_back("mosaic");
#endif
#ifdef PAIMON_ENABLE_ORC
values.emplace_back("orc");
#endif
#ifdef PAIMON_ENABLE_AVRO
values.emplace_back("avro");
#endif
return values;
}
INSTANTIATE_TEST_SUITE_P(FileFormat, BlobTableInteTest,
::testing::ValuesIn(GetTestValuesForBlobTableInteTest()));
TEST_P(BlobTableInteTest, TestAppendTableWriteWithBlobAsDescriptorTrue) {
arrow::FieldVector fields = {arrow::field("f0", arrow::utf8()),
arrow::field("f1", arrow::int32()),
BlobUtils::ToArrowField("blob", true)};
std::map<std::string, std::string> options = {{Options::FILE_FORMAT, GetParam()},
{Options::TARGET_FILE_SIZE, "700"},
{Options::BUCKET, "-1"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
{Options::BLOB_AS_DESCRIPTOR, "true"},
{Options::FILE_SYSTEM, "local"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
// prepare data: input uses plain raw blob bytes for readability
std::string raw_json = R"([
["str_0", null, "hello_blob_0"],
["str_1", 1, "blob_data_1"],
["str_2", 2, "blob_data_2"],
["str_3", null, "blob_data_3"]
])";
auto raw_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), raw_json).ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto desc_array, ConvertRawBlobToDescriptor(raw_array, {"blob"}));
// write descriptor array
auto schema = arrow::schema(fields);
ASSERT_OK_AND_ASSIGN(auto commit_msgs,
WriteArray(table_path, {}, schema->field_names(), {desc_array}));
ASSERT_OK(Commit(table_path, commit_msgs));
// read result contains descriptors pointing to paimon internal blob files
// resolve descriptors back to raw bytes, then prepend _VALUE_KIND and compare
ASSERT_OK_AND_ASSIGN(auto result, ScanAndReadResult(table_path, schema->field_names()));
ASSERT_TRUE(result);
auto read_concat = arrow::Concatenate(result->chunks()).ValueOrDie();
auto read_struct = std::dynamic_pointer_cast<arrow::StructArray>(read_concat);
ASSERT_OK_AND_ASSIGN(auto resolved, ConvertDescriptorToRawBlob(read_struct, {"blob"}));
ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(raw_array));
ASSERT_TRUE(resolved->Equals(expected_with_rk));
}
TEST_P(BlobTableInteTest, TestAppendTableWriteWithBlobAsDescriptorFalse) {
arrow::FieldVector fields = {arrow::field("f0", arrow::utf8()),
arrow::field("f1", arrow::int32()),
BlobUtils::ToArrowField("blob", true)};
std::map<std::string, std::string> options = {{Options::FILE_FORMAT, GetParam()},
{Options::TARGET_FILE_SIZE, "700"},
{Options::BUCKET, "-1"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
{Options::BLOB_AS_DESCRIPTOR, "false"},
{Options::FILE_SYSTEM, "local"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
std::string data_json = R"([
["str_0", null, "apple"],
["str_1", 1, "banana"],
["str_2", 2, "cat"],
["str_3", null, "dog"]
])";
auto write_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), data_json).ValueOrDie());
auto schema = arrow::schema(fields);
ASSERT_OK_AND_ASSIGN(auto commit_msgs,
WriteArray(table_path, {}, schema->field_names(), {write_array}));
ASSERT_OK(Commit(table_path, commit_msgs));
// BLOB_AS_DESCRIPTOR=false: blob data is stored inline, read result should match input
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), write_array));
}
TEST_P(BlobTableInteTest, TestWriteNullOnMissingFile) {
// blob-write-null-on-missing-file=true: a descriptor whose file is gone at write time
// becomes a NULL blob element, while the row itself is still written and merged with
// the data file on read.
arrow::FieldVector fields = {arrow::field("f0", arrow::utf8()),
arrow::field("f1", arrow::int32()),
BlobUtils::ToArrowField("blob", true)};
std::map<std::string, std::string> options = {
{Options::FILE_FORMAT, GetParam()},
{Options::TARGET_FILE_SIZE, "700"},
{Options::BUCKET, "-1"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
{Options::BLOB_AS_DESCRIPTOR, "true"},
{Options::BLOB_WRITE_NULL_ON_MISSING_FILE, "true"},
{Options::FILE_SYSTEM, "local"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
std::string raw_json = R"([
["str_0", 0, "blob_data_0"],
["str_1", 1, "blob_data_1"],
["str_2", 2, "blob_data_2"]
])";
auto raw_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), raw_json).ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto desc_array, ConvertRawBlobToDescriptor(raw_array, {"blob"}));
// Remove the file behind row 1's descriptor so the write observes a missing file.
ASSERT_OK(DeleteDescriptorTarget(desc_array, "blob", /*row=*/1));
auto schema = arrow::schema(fields);
ASSERT_OK_AND_ASSIGN(auto commit_msgs,
WriteArray(table_path, {}, schema->field_names(), {desc_array}));
ASSERT_OK(Commit(table_path, commit_msgs));
// Both the main data file and the .blob file keep the full row count; the read path
// merges them back into rows where row 1's blob is NULL.
ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
VerifyDataFileMetas(plan, /*expected_file_count=*/2, /*expected_row_counts=*/{3, 3},
/*expected_min_seqs=*/{1, 1}, /*expected_max_seqs=*/{1, 1},
/*expected_first_row_ids=*/{0, 0},
/*expected_write_cols=*/
{std::vector<std::string>{"f0", "f1"}, std::vector<std::string>{"blob"}});
ASSERT_OK_AND_ASSIGN(auto result, ReadTable(table_path, schema->field_names(), plan));
ASSERT_TRUE(result);
auto read_concat = arrow::Concatenate(result->chunks()).ValueOrDie();
auto read_struct = std::dynamic_pointer_cast<arrow::StructArray>(read_concat);
ASSERT_OK_AND_ASSIGN(auto resolved, ConvertDescriptorToRawBlob(read_struct, {"blob"}));
std::string expected_json = R"([
["str_0", 0, "blob_data_0"],
["str_1", 1, null],
["str_2", 2, "blob_data_2"]
])";
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), expected_json)
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(expected_array));
ASSERT_TRUE(resolved->Equals(expected_with_rk))
<< "result:" << resolved->ToString() << std::endl
<< "expected:" << expected_with_rk->ToString();
}
TEST_P(BlobTableInteTest, TestMissingFileFailsWriteWhenWriteNullDisabled) {
// Without blob-write-null-on-missing-file, a missing descriptor file fails the write.
arrow::FieldVector fields = {arrow::field("f0", arrow::utf8()),
arrow::field("f1", arrow::int32()),
BlobUtils::ToArrowField("blob", true)};
std::map<std::string, std::string> options = {{Options::FILE_FORMAT, GetParam()},
{Options::TARGET_FILE_SIZE, "700"},
{Options::BUCKET, "-1"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
{Options::BLOB_AS_DESCRIPTOR, "true"},
{Options::FILE_SYSTEM, "local"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
std::string raw_json = R"([
["str_0", 0, "blob_data_0"],
["str_1", 1, "blob_data_1"]
])";
auto raw_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), raw_json).ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto desc_array, ConvertRawBlobToDescriptor(raw_array, {"blob"}));
ASSERT_OK(DeleteDescriptorTarget(desc_array, "blob", /*row=*/1));
auto schema = arrow::schema(fields);
ASSERT_NOK_WITH_MSG(WriteArray(table_path, {}, schema->field_names(), {desc_array}),
"not exists");
}
TEST_P(BlobTableInteTest, TestWriteNullOnFetchFailure) {
// blob-write-null-on-fetch-failure=true: a descriptor whose data cannot be fetched for a
// reason other than a missing file (here: offset beyond the end of the file) becomes a
// NULL blob element, while the row itself is still written and merged with the data file
// on read.
arrow::FieldVector fields = {arrow::field("f0", arrow::utf8()),
arrow::field("f1", arrow::int32()),
BlobUtils::ToArrowField("blob", true)};
std::map<std::string, std::string> options = {
{Options::FILE_FORMAT, GetParam()},
{Options::TARGET_FILE_SIZE, "700"},
{Options::BUCKET, "-1"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
{Options::BLOB_AS_DESCRIPTOR, "true"},
{Options::BLOB_WRITE_NULL_ON_FETCH_FAILURE, "true"},
{Options::FILE_SYSTEM, "local"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
std::string raw_json = R"([
["str_0", 0, "blob_data_0"],
["str_1", 1, "blob_data_1"],
["str_2", 2, "blob_data_2"]
])";
auto raw_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), raw_json).ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto desc_array, ConvertRawBlobToDescriptor(raw_array, {"blob"}));
// Point row 1's descriptor at an offset beyond the end of its file so the write observes
// a fetch failure that is not a missing file.
ASSERT_OK_AND_ASSIGN(desc_array, CorruptDescriptorOffset(desc_array, "blob", /*row=*/1));
auto schema = arrow::schema(fields);
ASSERT_OK_AND_ASSIGN(auto commit_msgs,
WriteArray(table_path, {}, schema->field_names(), {desc_array}));
ASSERT_OK(Commit(table_path, commit_msgs));
// Both the main data file and the .blob file keep the full row count; the read path
// merges them back into rows where row 1's blob is NULL.
ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
VerifyDataFileMetas(plan, /*expected_file_count=*/2, /*expected_row_counts=*/{3, 3},
/*expected_min_seqs=*/{1, 1}, /*expected_max_seqs=*/{1, 1},
/*expected_first_row_ids=*/{0, 0},
/*expected_write_cols=*/
{std::vector<std::string>{"f0", "f1"}, std::vector<std::string>{"blob"}});
ASSERT_OK_AND_ASSIGN(auto result, ReadTable(table_path, schema->field_names(), plan));
ASSERT_TRUE(result);
auto read_concat = arrow::Concatenate(result->chunks()).ValueOrDie();
auto read_struct = std::dynamic_pointer_cast<arrow::StructArray>(read_concat);
ASSERT_OK_AND_ASSIGN(auto resolved, ConvertDescriptorToRawBlob(read_struct, {"blob"}));
std::string expected_json = R"([
["str_0", 0, "blob_data_0"],
["str_1", 1, null],
["str_2", 2, "blob_data_2"]
])";
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), expected_json)
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(expected_array));
ASSERT_TRUE(resolved->Equals(expected_with_rk))
<< "result:" << resolved->ToString() << std::endl
<< "expected:" << expected_with_rk->ToString();
}
TEST_P(BlobTableInteTest, TestWriteNullOnFetchFailureCoversMissingFile) {
// The existence check runs only under blob-write-null-on-missing-file, so with only
// blob-write-null-on-fetch-failure a missing file is converted as an ordinary fetch failure.
arrow::FieldVector fields = {arrow::field("f0", arrow::utf8()),
arrow::field("f1", arrow::int32()),
BlobUtils::ToArrowField("blob", true)};
std::map<std::string, std::string> options = {
{Options::FILE_FORMAT, GetParam()},
{Options::TARGET_FILE_SIZE, "700"},
{Options::BUCKET, "-1"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
{Options::BLOB_AS_DESCRIPTOR, "true"},
{Options::BLOB_WRITE_NULL_ON_FETCH_FAILURE, "true"},
{Options::FILE_SYSTEM, "local"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
std::string raw_json = R"([
["str_0", 0, "blob_data_0"],
["str_1", 1, "blob_data_1"],
["str_2", 2, "blob_data_2"]
])";
auto raw_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), raw_json).ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto desc_array, ConvertRawBlobToDescriptor(raw_array, {"blob"}));
ASSERT_OK(DeleteDescriptorTarget(desc_array, "blob", /*row=*/1));
auto schema = arrow::schema(fields);
ASSERT_OK_AND_ASSIGN(auto commit_msgs,
WriteArray(table_path, {}, schema->field_names(), {desc_array}));
ASSERT_OK(Commit(table_path, commit_msgs));
ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
VerifyDataFileMetas(plan, /*expected_file_count=*/2, /*expected_row_counts=*/{3, 3},
/*expected_min_seqs=*/{1, 1}, /*expected_max_seqs=*/{1, 1},
/*expected_first_row_ids=*/{0, 0},
/*expected_write_cols=*/
{std::vector<std::string>{"f0", "f1"}, std::vector<std::string>{"blob"}});
ASSERT_OK_AND_ASSIGN(auto result, ReadTable(table_path, schema->field_names(), plan));
ASSERT_TRUE(result);
auto read_concat = arrow::Concatenate(result->chunks()).ValueOrDie();
auto read_struct = std::dynamic_pointer_cast<arrow::StructArray>(read_concat);
ASSERT_OK_AND_ASSIGN(auto resolved, ConvertDescriptorToRawBlob(read_struct, {"blob"}));
std::string expected_json = R"([
["str_0", 0, "blob_data_0"],
["str_1", 1, null],
["str_2", 2, "blob_data_2"]
])";
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), expected_json)
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(expected_array));
ASSERT_TRUE(resolved->Equals(expected_with_rk))
<< "result:" << resolved->ToString() << std::endl
<< "expected:" << expected_with_rk->ToString();
}
TEST_P(BlobTableInteTest, TestBasic) {
CreateTable();
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields_);
// write field: f0, f1, f2
std::vector<std::string> write_cols0 = schema->field_names();
auto src_array0 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
[1, "a", "b"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs, WriteArray(table_path, {}, write_cols0, {src_array0}));
ASSERT_OK(Commit(table_path, commit_msgs));
// write field: f1, f2
std::vector<std::string> write_cols1 = {"f1", "f2"};
auto src_array1 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[1], fields_[2]}), R"([
["new_blob", "c"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(commit_msgs, WriteArray(table_path, {}, write_cols1, {src_array1}));
SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs);
ASSERT_OK(Commit(table_path, commit_msgs));
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
[1, "new_blob", "c"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
// read with row tracking
auto expected_row_tracking_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(
arrow::struct_({fields_[1], fields_[0], SpecialFields::SequenceNumber().ArrowField(),
SpecialFields::RowId().ArrowField(), fields_[2]}),
R"([
["new_blob", 1, 2, 0, "c"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, {"f1", "f0", "_SEQUENCE_NUMBER", "_ROW_ID", "f2"},
expected_row_tracking_array));
}
TEST_P(BlobTableInteTest, TestBlobFilesAcrossSchemaIds) {
std::map<std::string, std::string> options = {{Options::FILE_FORMAT, GetParam()},
{Options::FILE_SYSTEM, "local"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"}};
CreateTable(/*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
// Simulate the post-compaction layout: one normal file bridges blob files across schema ids.
std::vector<std::string> write_cols0 = {"f0", "f2"};
auto src_array0 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[0], fields_[2]}), R"([
[1, "a"],
[2, "b"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs0, WriteArray(table_path, {}, write_cols0, {src_array0}));
ASSERT_OK(Commit(table_path, commit_msgs0));
std::vector<std::string> blob_write_cols = {"f1"};
auto src_array1 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[1]}), R"([
["c"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs1,
WriteArray(table_path, {}, blob_write_cols, {src_array1}));
SetFirstRowId(0, commit_msgs1);
ASSERT_OK(Commit(table_path, commit_msgs1));
auto f3 = arrow::field("f3", arrow::int64());
ASSERT_OK(WriteNextSchema(table_path,
{DataField(0, fields_[0]), DataField(1, fields_[1]),
DataField(2, fields_[2]), DataField(3, f3)},
/*highest_field_id=*/3, options));
auto src_array2 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[1]}), R"([
["d"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs2,
WriteArray(table_path, {}, blob_write_cols, {src_array2}));
SetFirstRowId(1, commit_msgs2);
ASSERT_OK(Commit(table_path, commit_msgs2));
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
[1, "c", "a"],
[2, "d", "b"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, arrow::schema(fields_)->field_names(), expected_array));
}
TEST_P(BlobTableInteTest, TestMultipleAppends) {
CreateTable();
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields_);
// write field: f0, f1, f2
std::vector<std::string> write_cols0 = schema->field_names();
auto src_array0 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
[1, "a", "b"],
[1, "a", "b"],
[1, "a", "b"],
[1, "a", "b"],
[1, "a", "b"],
[1, "a", "b"],
[1, "a", "b"],
[1, "a", "b"],
[1, "a", "b"],
[1, "a", "b"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs, WriteArray(table_path, {}, write_cols0, {src_array0}));
ASSERT_OK(Commit(table_path, commit_msgs));
// write field: f0, f1
std::vector<std::string> write_cols1 = {"f0", "f1"};
auto src_array1 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[0], fields_[1]}), R"([
[1, "a"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, {}, write_cols1, {src_array1}));
SetFirstRowId(10, commit_msgs1);
// write field: f2
std::vector<std::string> write_cols2 = {"f2"};
auto src_array2 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[2]}), R"([
["b"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs2, WriteArray(table_path, {}, write_cols2, {src_array2}));
SetFirstRowId(10, commit_msgs2);
std::vector<std::shared_ptr<CommitMessage>> total_msgs;
total_msgs.insert(total_msgs.end(), commit_msgs1.begin(), commit_msgs1.end());
total_msgs.insert(total_msgs.end(), commit_msgs2.begin(), commit_msgs2.end());
ASSERT_OK(Commit(table_path, total_msgs));
// write field: f0, f1
std::vector<std::string> write_cols3 = {"f0", "f1"};
auto src_array3 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[0], fields_[1]}), R"([
[2, "c"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs3, WriteArray(table_path, {}, write_cols3, {src_array3}));
SetFirstRowId(11, commit_msgs3);
ASSERT_OK(Commit(table_path, commit_msgs3));
// write field: f2
std::vector<std::string> write_cols4 = {"f2"};
auto src_array4 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[2]}), R"([
["d"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs4, WriteArray(table_path, {}, write_cols4, {src_array4}));
SetFirstRowId(11, commit_msgs4);
ASSERT_OK(Commit(table_path, commit_msgs4));
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
[1, "a", "b"],
[1, "a", "b"],
[1, "a", "b"],
[1, "a", "b"],
[1, "a", "b"],
[1, "a", "b"],
[1, "a", "b"],
[1, "a", "b"],
[1, "a", "b"],
[1, "a", "b"],
[1, "a", "b"],
[2, "c", "d"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
// read with row tracking
auto expected_row_tracking_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({
fields_[0],
fields_[1],
fields_[2],
SpecialFields::RowId().ArrowField(),
SpecialFields::SequenceNumber().ArrowField(),
}),
R"([
[1, "a", "b", 0, 1],
[1, "a", "b", 1, 1],
[1, "a", "b", 2, 1],
[1, "a", "b", 3, 1],
[1, "a", "b", 4, 1],
[1, "a", "b", 5, 1],
[1, "a", "b", 6, 1],
[1, "a", "b", 7, 1],
[1, "a", "b", 8, 1],
[1, "a", "b", 9, 1],
[1, "a", "b", 10, 2],
[2, "c", "d", 11, 4]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, {"f0", "f1", "f2", "_ROW_ID", "_SEQUENCE_NUMBER"},
expected_row_tracking_array));
}
TEST_P(BlobTableInteTest, TestDataEvolutionBlobOnlyWriteWithFirstRowId) {
arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
arrow::field("f1", arrow::utf8()), BlobUtils::ToArrowField("b0")};
std::map<std::string, std::string> options = {{Options::FILE_FORMAT, GetParam()},
{Options::FILE_SYSTEM, "local"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields);
// Initial full-row write assigns row ids 0 and 1.
auto src_array0 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
[1, "a", "old_blob_0"],
[2, "b", "old_blob_1"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs0,
WriteArray(table_path, {}, schema->field_names(), {src_array0}));
ASSERT_OK(Commit(table_path, commit_msgs0));
// Update only b0 and align it with the existing row ids.
std::vector<std::string> blob_write_cols = {"b0"};
auto src_array1 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[2]}), R"([
["new_blob_0"],
["new_blob_1"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs1,
WriteArray(table_path, {}, blob_write_cols, {src_array1}));
SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs1);
ASSERT_OK(Commit(table_path, commit_msgs1));
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
[1, "a", "new_blob_0"],
[2, "b", "new_blob_1"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
auto expected_with_row_id = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(
arrow::struct_({fields[0], fields[1], fields[2], SpecialFields::RowId().ArrowField()}),
R"([
[1, "a", "new_blob_0", 0],
[2, "b", "new_blob_1", 1]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, {"f0", "f1", "b0", "_ROW_ID"}, expected_with_row_id));
}
/// Build a single-blob-column StructArray for a data-evolution partial update: "PH" marks a row
/// whose blob is not updated (persisted as a placeholder entry), std::nullopt a null blob.
std::shared_ptr<arrow::StructArray> MakeBlobUpdateArray(
const std::shared_ptr<arrow::Field>& blob_field,
const std::vector<std::optional<std::string>>& rows) {
auto struct_type = arrow::struct_({blob_field});
arrow::StructBuilder struct_builder(struct_type, arrow::default_memory_pool(),
{std::make_shared<arrow::LargeBinaryBuilder>()});
auto blob_builder = checked_cast<arrow::LargeBinaryBuilder*>(struct_builder.field_builder(0));
for (const auto& row : rows) {
EXPECT_TRUE(struct_builder.Append().ok());
if (!row) {
EXPECT_TRUE(blob_builder->AppendNull().ok());
} else if (*row == "PH") {
std::string_view sentinel = BlobDefs::PlaceholderSentinelView();
EXPECT_TRUE(blob_builder->Append(sentinel.data(), sentinel.size()).ok());
} else {
EXPECT_TRUE(blob_builder->Append(row->data(), row->size()).ok());
}
}
std::shared_ptr<arrow::Array> array;
EXPECT_TRUE(struct_builder.Finish(&array).ok());
return std::dynamic_pointer_cast<arrow::StructArray>(array);
}
TEST_P(BlobTableInteTest, TestDataEvolutionBlobPartialUpdateFallback) {
// the blob column is updated to null below, so it must be nullable
arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
arrow::field("f1", arrow::utf8()),
BlobUtils::ToArrowField("b0", /*nullable=*/true)};
std::map<std::string, std::string> options = {{Options::FILE_FORMAT, GetParam()},
{Options::FILE_SYSTEM, "local"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields);
// Initial full-row write assigns row ids 0-2.
auto src_array0 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
[1, "a", "blob_a"],
[2, "b", "blob_b"],
[3, "c", "blob_c"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs0,
WriteArray(table_path, {}, schema->field_names(), {src_array0}));
ASSERT_OK(Commit(table_path, commit_msgs0));
// Partial update: only row 1 gets a new blob, the untouched rows are written as
// placeholder entries and must fall back to the previous blob file when read.
auto update_array = MakeBlobUpdateArray(fields[2], {"PH", "updated_b", "PH"});
ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, {}, {"b0"}, {update_array}));
SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs1);
ASSERT_OK(Commit(table_path, commit_msgs1));
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
[1, "a", "blob_a"],
[2, "b", "updated_b"],
[3, "c", "blob_c"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
// updating a blob to null is not a placeholder: the null must win over older layers
auto null_update_array = MakeBlobUpdateArray(fields[2], {std::nullopt, "PH", "PH"});
ASSERT_OK_AND_ASSIGN(auto commit_msgs2,
WriteArray(table_path, {}, {"b0"}, {null_update_array}));
SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs2);
ASSERT_OK(Commit(table_path, commit_msgs2));
auto expected_array2 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
[1, "a", null],
[2, "b", "updated_b"],
[3, "c", "blob_c"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array2));
// row ids still come from the data files and stay aligned with the fallback result
auto expected_with_row_id = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(
arrow::struct_({fields[0], fields[1], fields[2], SpecialFields::RowId().ArrowField()}),
R"([
[1, "a", null, 0],
[2, "b", "updated_b", 1],
[3, "c", "blob_c", 2]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, {"f0", "f1", "b0", "_ROW_ID"}, expected_with_row_id));
// blob_as_descriptor read mode: the fallback-merged values come back as descriptors and
// must still resolve to the same bytes
ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
std::map<std::string, std::string> read_options = {{Options::BLOB_AS_DESCRIPTOR, "true"}};
ASSERT_OK_AND_ASSIGN(auto desc_result, ReadTable(table_path, schema->field_names(), plan,
/*predicate=*/nullptr, read_options));
ASSERT_TRUE(desc_result);
auto desc_concat = arrow::Concatenate(desc_result->chunks()).ValueOrDie();
auto desc_struct = std::dynamic_pointer_cast<arrow::StructArray>(desc_concat);
ASSERT_TRUE(desc_struct);
ASSERT_OK_AND_ASSIGN(auto resolved, ConvertDescriptorToRawBlob(desc_struct, {"b0"}));
ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(expected_array2));
ASSERT_TRUE(resolved->Equals(expected_with_rk))
<< "result:" << resolved->ToString() << "\nexpected:" << expected_with_rk->ToString();
}
TEST_P(BlobTableInteTest, TestDataEvolutionPartialUpdateWithNonBlobColumnFallsBack) {
arrow::FieldVector fields = {
arrow::field("f0", arrow::int32()), arrow::field("f1", arrow::utf8()),
BlobUtils::ToArrowField("b0", /*nullable=*/true),
arrow::field("a0", arrow::list(BlobUtils::ToArrowField("item", /*nullable=*/true)))};
std::map<std::string, std::string> options = {{Options::FILE_FORMAT, GetParam()},
{Options::FILE_SYSTEM, "local"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields);
auto src_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
[1, "a", "blob_a", ["a0", "a1"]],
[2, "b", "blob_b", ["b0"]],
[3, "c", "blob_c", []]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs0,
WriteArray(table_path, {}, schema->field_names(), {src_array}));
ASSERT_OK(Commit(table_path, commit_msgs0));
std::vector<std::string> update_cols = {"f0", "b0", "a0"};
auto update_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[0], fields[2], fields[3]}),
R"([
[10, "_PAIMON_BLOB_PLACEHOLDER", ["_PAIMON_BLOB_PLACEHOLDER"]],
[20, "updated_b", ["updated_b0", null]],
[30, "_PAIMON_BLOB_PLACEHOLDER", ["_PAIMON_BLOB_PLACEHOLDER"]]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs1,
WriteArray(table_path, {}, update_cols, {update_array}));
SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs1);
ASSERT_OK(Commit(table_path, commit_msgs1));
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
[10, "a", "blob_a", ["a0", "a1"]],
[20, "b", "updated_b", ["updated_b0", null]],
[30, "c", "blob_c", []]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
auto expected_with_row_id = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(
arrow::struct_({fields[0], fields[2], fields[3], SpecialFields::RowId().ArrowField()}),
R"([
[10, "blob_a", ["a0", "a1"], 0],
[20, "updated_b", ["updated_b0", null], 1],
[30, "blob_c", [], 2]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, {"f0", "b0", "a0", "_ROW_ID"}, expected_with_row_id));
}
TEST_P(BlobTableInteTest, TestArrayBlobWriteAndPartialUpdate) {
arrow::FieldVector fields = {
arrow::field("f0", arrow::int32()),
arrow::field("a0", arrow::list(BlobUtils::ToArrowField("item", /*nullable=*/true)))};
std::map<std::string, std::string> options = {
{Options::FILE_FORMAT, GetParam()},
{Options::FILE_SYSTEM, "local"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
{Options::BLOB_WRITE_NULL_ON_FETCH_FAILURE, "true"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields);
auto src_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
[1, ["a", null, ""]],
[2, null],
[3, []],
[4, ["b"]]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs0,
WriteArray(table_path, {}, schema->field_names(), {src_array}));
ASSERT_OK(Commit(table_path, commit_msgs0));
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), src_array));
auto update_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[1]}), R"([
[["_PAIMON_BLOB_PLACEHOLDER"]],
[["updated", null]],
[null],
[["_PAIMON_BLOB_PLACEHOLDER"]]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, {}, {"a0"}, {update_array}));
ASSERT_NOK_WITH_MSG(Commit(table_path, commit_msgs1),
"blobStart 4 should be less than start 4");
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), src_array));
SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs1);
ASSERT_OK(Commit(table_path, commit_msgs1));
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
[1, ["a", null, ""]],
[2, ["updated", null]],
[3, null],
[4, ["b"]]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
ASSERT_OK_AND_ASSIGN(std::unique_ptr<Blob> missing_blob,
Blob::FromPath(blob_dir_->Str() + "/missing.bin"));
PAIMON_UNIQUE_PTR<Bytes> missing_descriptor = missing_blob->ToDescriptor(pool_);
const std::string sentinel(BlobDefs::PlaceholderSentinelView());
ASSERT_OK_AND_ASSIGN(
std::shared_ptr<arrow::Array> update_list,
MakeArrayBlobArray(
fields[1]->type(),
{BlobElements{std::string(missing_descriptor->data(), missing_descriptor->size()), "x"},
BlobElements{sentinel}, BlobElements{sentinel}, BlobElements{sentinel}}));
auto second_update_array = arrow::StructArray::Make({update_list}, {fields[1]}).ValueOrDie();
ASSERT_OK_AND_ASSIGN(auto commit_msgs2,
WriteArray(table_path, {}, {"a0"}, {second_update_array}));
SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs2);
ASSERT_OK(Commit(table_path, commit_msgs2));
expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
[1, [null, "x"]],
[2, ["updated", null]],
[3, null],
[4, ["b"]]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
}
TEST_P(BlobTableInteTest, TestArrayBlobWriteDescriptorElements) {
auto element_field = BlobUtils::ToArrowField("item", /*nullable=*/true);
arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
arrow::field("a0", arrow::list(element_field))};
std::map<std::string, std::string> options = {{Options::FILE_FORMAT, GetParam()},
{Options::FILE_SYSTEM, "local"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields);
auto raw_elements = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({element_field}),
R"([["alpha"], ["omega"], ["single"]])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto descriptor_elements,
ConvertRawBlobToDescriptor(raw_elements, {"item"}));
const auto& descriptors =
checked_cast<const arrow::LargeBinaryArray&>(*descriptor_elements->field(0));
ASSERT_OK_AND_ASSIGN(
std::shared_ptr<arrow::Array> list_array,
MakeArrayBlobArray(fields[1]->type(),
{BlobElements{std::string(descriptors.GetView(0)), std::nullopt,
"inline", std::string(descriptors.GetView(1))},
std::nullopt, BlobElements{std::string(descriptors.GetView(2))}}));
auto f0_array =
arrow::ipc::internal::json::ArrayFromJSON(arrow::int32(), "[1, 2, 3]").ValueOrDie();
auto write_array = arrow::StructArray::Make({f0_array, list_array}, fields).ValueOrDie();
ASSERT_OK_AND_ASSIGN(auto commit_msgs,
WriteArray(table_path, {}, schema->field_names(), {write_array}));
ASSERT_OK(Commit(table_path, commit_msgs));
blob_dir_.reset();
const std::string expected_json = R"([
[1, ["alpha", null, "inline", "omega"]],
[2, null],
[3, ["single"]]
])";
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), expected_json)
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
std::map<std::string, std::string> read_options = {{Options::BLOB_AS_DESCRIPTOR, "true"}};
ASSERT_OK_AND_ASSIGN(auto result, ReadTable(table_path, schema->field_names(), plan,
/*predicate=*/nullptr, read_options));
ASSERT_TRUE(result);
auto rows = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::Concatenate(result->chunks()).ValueOrDie());
ASSERT_TRUE(rows);
CheckArrayBlobColumn(rows, "a0", R"([["alpha", null, "inline", "omega"], null, ["single"]])",
/*blob_as_descriptor=*/true);
}
TEST_P(BlobTableInteTest, TestArrayBlobWithBlobFieldAcrossMultipleBlobFiles) {
arrow::FieldVector fields = {
arrow::field("f0", arrow::int32()), BlobUtils::ToArrowField("b0", /*nullable=*/true),
arrow::field("a0", arrow::list(BlobUtils::ToArrowField("item", /*nullable=*/true)))};
std::map<std::string, std::string> options = {{Options::FILE_FORMAT, GetParam()},
{Options::BLOB_TARGET_FILE_SIZE, "80"},
{Options::BUCKET, "-1"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
{Options::FILE_SYSTEM, "local"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields);
auto src_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
[0, "b0", ["x0", "y0"]],
[1, "b1", null],
[2, null, []],
[3, "b3", ["x3", null]],
[4, "b4", ["x4"]],
[5, "b5", ["x5", "y5", "z5"]],
[6, "b6", [null]],
[7, "b7", ["x7"]]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs,
WriteArray(table_path, {}, schema->field_names(), {src_array}));
ASSERT_OK(Commit(table_path, commit_msgs));
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), src_array));
ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
size_t blob_file_count = 0;
size_t array_blob_file_count = 0;
for (const auto& split : plan->Splits()) {
auto data_split = std::dynamic_pointer_cast<DataSplitImpl>(split);
ASSERT_TRUE(data_split);
for (const auto& file : data_split->DataFiles()) {
if (!BlobUtils::IsBlobFile(file->file_name)) {
continue;
}
if (file->write_cols == std::optional<std::vector<std::string>>({"a0"})) {
++array_blob_file_count;
} else {
ASSERT_EQ(file->write_cols, std::optional<std::vector<std::string>>({"b0"}));
++blob_file_count;
}
}
}
ASSERT_GT(blob_file_count, 0u);
ASSERT_GT(array_blob_file_count, 1u);
{
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
[1, "b1", null],
[5, "b5", ["x5", "y5", "z5"]],
[7, "b7", ["x7"]]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array,
/*predicate=*/nullptr,
/*row_ranges=*/{Range(1l, 1l), Range(5l, 5l), Range(7l, 7l)}));
}
{
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[2]}), R"([
[[]],
[["x3", null]],
[[null]]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, {"a0"}, expected_array,
/*predicate=*/nullptr,
/*row_ranges=*/{Range(2l, 3l), Range(6l, 6l)}));
}
{
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[0], fields[1]}), R"([
[0, "b0"], [1, "b1"], [2, null], [3, "b3"], [4, "b4"], [5, "b5"], [6, "b6"], [7, "b7"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, {"f0", "b0"}, expected_array));
}
const std::string updated_row_4(60, 'u');
const std::string updated_row_6(60, 'v');
const std::string sentinel(BlobDefs::PlaceholderSentinelView());
ASSERT_OK_AND_ASSIGN(
std::shared_ptr<arrow::Array> update_list,
MakeArrayBlobArray(fields[2]->type(),
{BlobElements{updated_row_4}, BlobElements{sentinel},
BlobElements{updated_row_6, std::nullopt}, std::nullopt}));
auto update_array = arrow::StructArray::Make({update_list}, {fields[2]}).ValueOrDie();
ASSERT_OK_AND_ASSIGN(auto update_msgs, WriteArray(table_path, {}, {"a0"}, {update_array}));
int64_t next_row_id = 4;
size_t update_file_count = 0;
for (const auto& commit_msg : update_msgs) {
auto commit_msg_impl = std::dynamic_pointer_cast<CommitMessageImpl>(commit_msg);
ASSERT_TRUE(commit_msg_impl);
for (const auto& file : commit_msg_impl->GetNewFilesIncrement().NewFiles()) {
file->AssignFirstRowId(next_row_id);
next_row_id += file->row_count;
++update_file_count;
}
}
ASSERT_EQ(next_row_id, 8);
ASSERT_GT(update_file_count, 1u);
ASSERT_OK(Commit(table_path, update_msgs));
ASSERT_OK_AND_ASSIGN(
std::shared_ptr<arrow::Array> expected_list,
MakeArrayBlobArray(fields[2]->type(),
{BlobElements{"x0", "y0"}, std::nullopt, BlobElements{},
BlobElements{"x3", std::nullopt}, BlobElements{updated_row_4},
BlobElements{"x5", "y5", "z5"},
BlobElements{updated_row_6, std::nullopt}, std::nullopt}));
auto expected_array =
arrow::StructArray::Make({src_array->field(0), src_array->field(1), expected_list}, fields)
.ValueOrDie();
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
}
TEST_P(BlobTableInteTest, TestMapBlobWriteAndPartialUpdate) {
arrow::FieldVector fields = {
arrow::field("f0", arrow::int32()),
arrow::field(
"m0", arrow::map(arrow::utf8(), BlobUtils::ToArrowField("value", /*nullable=*/true)))};
std::map<std::string, std::string> options = {
{Options::FILE_FORMAT, GetParam()},
{Options::FILE_SYSTEM, "local"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
{Options::BLOB_WRITE_NULL_ON_FETCH_FAILURE, "true"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields);
auto src_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
[1, [["a", "va"], ["", "empty key"], ["null", null]]],
[2, null],
[3, []],
[4, [["b", "vb"]]]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs0,
WriteArray(table_path, {}, schema->field_names(), {src_array}));
ASSERT_OK(Commit(table_path, commit_msgs0));
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), src_array));
auto update_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[1]}), R"([
[[["k", null], ["k", null]]],
[[["updated", "vu"], ["null", null]]],
[null],
[[["k", null], ["k", null]]]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, {}, {"m0"}, {update_array}));
SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs1);
ASSERT_OK(Commit(table_path, commit_msgs1));
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
[1, [["a", "va"], ["", "empty key"], ["null", null]]],
[2, [["updated", "vu"], ["null", null]]],
[3, null],
[4, [["b", "vb"]]]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
const std::string alpha_path = blob_dir_->Str() + "/alpha.bin";
ASSERT_OK(std::make_shared<LocalFileSystem>()->WriteFile(alpha_path, "alpha",
/*overwrite=*/true));
ASSERT_OK_AND_ASSIGN(std::unique_ptr<Blob> alpha_blob, Blob::FromPath(alpha_path));
PAIMON_UNIQUE_PTR<Bytes> alpha_descriptor = alpha_blob->ToDescriptor(pool_);
ASSERT_OK_AND_ASSIGN(std::unique_ptr<Blob> missing_blob,
Blob::FromPath(blob_dir_->Str() + "/missing.bin"));
PAIMON_UNIQUE_PTR<Bytes> missing_descriptor = missing_blob->ToDescriptor(pool_);
const MapBlobEntries placeholder = {{"k", std::nullopt}, {"k", std::nullopt}};
ASSERT_OK_AND_ASSIGN(
std::shared_ptr<arrow::Array> update_map,
MakeMapBlobArray(
fields[1]->type(),
{MapBlobEntries{
{"missing", std::string(missing_descriptor->data(), missing_descriptor->size())},
{"desc", std::string(alpha_descriptor->data(), alpha_descriptor->size())}},
placeholder, placeholder, placeholder}));
auto second_update_array = arrow::StructArray::Make({update_map}, {fields[1]}).ValueOrDie();
ASSERT_OK_AND_ASSIGN(auto commit_msgs2,
WriteArray(table_path, {}, {"m0"}, {second_update_array}));
SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs2);
ASSERT_OK(Commit(table_path, commit_msgs2));
blob_dir_.reset();
expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
[1, [["missing", null], ["desc", "alpha"]]],
[2, [["updated", "vu"], ["null", null]]],
[3, null],
[4, [["b", "vb"]]]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
std::map<std::string, std::string> read_options = {{Options::BLOB_AS_DESCRIPTOR, "true"}};
ASSERT_OK_AND_ASSIGN(auto result, ReadTable(table_path, schema->field_names(), plan,
/*predicate=*/nullptr, read_options));
ASSERT_TRUE(result);
auto rows = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::Concatenate(result->chunks()).ValueOrDie());
ASSERT_TRUE(rows);
CheckMapBlobColumn(rows, "m0",
R"([[["missing", null], ["desc", "alpha"]],
[["updated", "vu"], ["null", null]], null, [["b", "vb"]]])",
/*blob_as_descriptor=*/true);
}
TEST_P(BlobTableInteTest, TestDataEvolutionBlobPartialUpdateMultipleLayers) {
arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
arrow::field("f1", arrow::utf8()), BlobUtils::ToArrowField("b0")};
std::map<std::string, std::string> options = {{Options::FILE_FORMAT, GetParam()},
{Options::FILE_SYSTEM, "local"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields);
auto src_array0 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
[1, "a", "blob_0"],
[2, "b", "blob_1"],
[3, "c", "blob_2"],
[4, "d", "blob_3"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs0,
WriteArray(table_path, {}, schema->field_names(), {src_array0}));
ASSERT_OK(Commit(table_path, commit_msgs0));
// second layer updates row 0 within rows [0, 1]
auto update_array1 = MakeBlobUpdateArray(fields[2], {"update1_0", "PH"});
ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, {}, {"b0"}, {update_array1}));
SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs1);
ASSERT_OK(Commit(table_path, commit_msgs1));
// third layer updates row 3 within rows [2, 3]; each layer only partially covers the
// range, the uncovered parts behave as placeholders
auto update_array2 = MakeBlobUpdateArray(fields[2], {"PH", "update2_3"});
ASSERT_OK_AND_ASSIGN(auto commit_msgs2, WriteArray(table_path, {}, {"b0"}, {update_array2}));
SetFirstRowId(/*reset_first_row_id=*/2, commit_msgs2);
ASSERT_OK(Commit(table_path, commit_msgs2));
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
[1, "a", "update1_0"],
[2, "b", "blob_1"],
[3, "c", "blob_2"],
[4, "d", "update2_3"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
// row-range pushdown where the selected rows fall in one layer's file and in another
// layer's uncovered gap: row 1 is a gap row for the third layer, row 2 for the second
auto expected_middle = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
[2, "b", "blob_1"],
[3, "c", "blob_2"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_middle,
/*predicate=*/nullptr, /*row_ranges=*/{Range(1, 2)}));
// per-row reads resolve every row independently
const std::vector<std::string> expected_rows = {
R"([[1, "a", "update1_0"]])", R"([[2, "b", "blob_1"]])", R"([[3, "c", "blob_2"]])",
R"([[4, "d", "update2_3"]])"};
ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> unpruned_plan, ScanTable(table_path));
ASSERT_EQ(unpruned_plan->Splits().size(), 1);
auto unpruned_split = std::dynamic_pointer_cast<DataSplitImpl>(unpruned_plan->Splits()[0]);
ASSERT_NE(unpruned_split, nullptr);
ASSERT_EQ(unpruned_split->DataFiles().size(), 4);
for (int32_t i = 0; i < static_cast<int32_t>(expected_rows.size()); i++) {
SCOPED_TRACE(i);
auto expected_single = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), expected_rows[i])
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_single,
/*predicate=*/nullptr, /*row_ranges=*/{Range(i, i)}));
// Range planning can narrow a read after scanning, leaving unselected BLOB files
// in the split. Their sequence layers must still align with the selected rows.
auto indexed_split =
std::make_shared<IndexedSplitImpl>(unpruned_split, std::vector<Range>{Range(i, i)});
auto indexed_plan = std::make_shared<PlanImpl>(
unpruned_plan->SnapshotId(), std::vector<std::shared_ptr<Split>>{indexed_split});
ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> result,
ReadTable(table_path, schema->field_names(), indexed_plan));
ASSERT_NE(result, nullptr);
ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::StructArray> expected_with_row_kind,
PrependRowKindColumn(expected_single));
ASSERT_TRUE(result->Equals(arrow::ChunkedArray(expected_with_row_kind)))
<< result->ToString();
}
}
TEST_P(BlobTableInteTest, TestDataEvolutionBlobPartialUpdateWithDeletionVectors) {
arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
arrow::field("f1", arrow::utf8()), BlobUtils::ToArrowField("b0")};
std::map<std::string, std::string> options = {{Options::FILE_FORMAT, GetParam()},
{Options::FILE_SYSTEM, "local"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
{Options::DELETION_VECTORS_ENABLED, "true"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields);
auto src_array0 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
[1, "a", "blob_0"],
[2, "b", "blob_1"],
[3, "c", "blob_2"],
[4, "d", "blob_3"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs0,
WriteArray(table_path, {}, schema->field_names(), {src_array0}));
SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs0);
ASSERT_OK(Commit(table_path, commit_msgs0));
// second layer updates row 0 within rows [0, 1], rows [2, 3] are its placeholder gap
auto update_array1 = MakeBlobUpdateArray(fields[2], {"update1_0", "PH"});
ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, {}, {"b0"}, {update_array1}));
SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs1);
ASSERT_OK(Commit(table_path, commit_msgs1));
// third layer updates row 3 within rows [2, 3], rows [0, 1] are its placeholder gap
auto update_array2 = MakeBlobUpdateArray(fields[2], {"PH", "update2_3"});
ASSERT_OK_AND_ASSIGN(auto commit_msgs2, WriteArray(table_path, {}, {"b0"}, {update_array2}));
SetFirstRowId(/*reset_first_row_id=*/2, commit_msgs2);
ASSERT_OK(Commit(table_path, commit_msgs2));
// delete row 1 and row 3: each sits inside one blob layer's file and the other layer's
// gap, so both blob layers and the normal file must drop the same rows to stay aligned
ASSERT_OK_AND_ASSIGN(std::string anchor_file_name,
DeletionVectorTestHelper::RetrieveAnchorFileName(commit_msgs0));
ASSERT_OK(CommitDeletionVectors(table_path, commit_msgs0[0],
{{anchor_file_name, /*deleted_positions=*/{1, 3}}}));
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
[1, "a", "update1_0"],
[3, "c", "blob_2"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
auto expected_with_row_id = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(
arrow::struct_({fields[0], fields[1], fields[2], SpecialFields::RowId().ArrowField()}),
R"([
[1, "a", "update1_0", 0],
[3, "c", "blob_2", 2]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, {"f0", "f1", "b0", "_ROW_ID"}, expected_with_row_id));
// Projecting only the blob column leaves no data bunch, so the whole read is the blob
// fallback path: the layer files carry the shifted deletion vector and the placeholder gap
// segments drop the deleted row ids, with no normal file to anchor the row count.
auto expected_blob_only = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[2]}), R"([
["update1_0"],
["blob_2"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, {"b0"}, expected_blob_only));
// a row-range selection composes with the deletion vector: rows {0, 1} minus deleted {1}.
// The range stops before row 2, the other survivor, so it cannot pass by returning what the
// unrestricted read above already returned.
auto expected_selected = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
[1, "a", "update1_0"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_selected,
/*predicate=*/nullptr, /*row_ranges=*/{Range(0, 1)}));
}
TEST_P(BlobTableInteTest, TestDataEvolutionPartialBlobBackfillWithDeletionVectors) {
arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
BlobUtils::ToArrowField("b0", /*nullable=*/true)};
CreateTable(fields, {},
{{Options::FILE_FORMAT, GetParam()},
{Options::FILE_SYSTEM, "local"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
{Options::DELETION_VECTORS_ENABLED, "true"},
{Options::DELETION_VECTOR_BITMAP64, "true"},
{Options::READ_BATCH_SIZE, "2"}});
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto normal_type = arrow::struct_({fields[0]});
auto normal = arrow::ipc::internal::json::ArrayFromJSON(
normal_type, R"([[0],[1],[2],[3],[4],[5],[6],[7],[8],[9]])")
.ValueOrDie();
ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> base,
WriteArray(table_path, {}, {"f0"}, {normal}));
SetFirstRowId(100, base);
ASSERT_OK(Commit(table_path, base));
auto blobs = MakeBlobUpdateArray(fields[1], {"b3", "b4", "b5"});
ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> backfill,
WriteArray(table_path, {}, {"b0"}, {blobs}));
SetFirstRowId(103, backfill);
ASSERT_OK(Commit(table_path, backfill));
auto full = checked_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields),
R"([[0,null],[1,null],[2,null],[3,"b3"],[4,"b4"],
[5,"b5"],[6,null],[7,null],[8,null],[9,null]])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, {"f0", "b0"}, full));
auto blob_only = checked_pointer_cast<arrow::StructArray>(
arrow::StructArray::Make({full->field(1)}, {fields[1]}).ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, {"b0"}, blob_only));
ASSERT_OK_AND_ASSIGN(std::string anchor,
DeletionVectorTestHelper::RetrieveAnchorFileName(base));
// Delete rows in the physical blob file, the leading gap, and the trailing gap.
ASSERT_OK(CommitDeletionVectors(table_path, base[0], {{anchor, {1, 3, 7, 9}}}));
arrow::FieldVector read_fields = {fields[0], fields[1], SpecialFields::RowId().ArrowField()};
auto expected = checked_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(
arrow::struct_(read_fields),
R"([[0,null,100],[2,null,102],[4,"b4",104],[5,"b5",105],[6,null,106],[8,null,108]])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, {"f0", "b0", "_ROW_ID"}, expected));
blob_only = checked_pointer_cast<arrow::StructArray>(
arrow::StructArray::Make({expected->field(1)}, {fields[1]}).ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, {"b0"}, blob_only));
auto selected = checked_pointer_cast<arrow::StructArray>(blob_only->Slice(1, 4));
ASSERT_OK(ScanAndRead(table_path, {"b0"}, selected, nullptr, {Range(102, 106)}));
auto gap_only = checked_pointer_cast<arrow::StructArray>(blob_only->Slice(4, 2));
ASSERT_OK(ScanAndRead(table_path, {"b0"}, gap_only, nullptr, {Range(106, 108)}));
ASSERT_OK(ScanAndRead(table_path, {"b0"}, nullptr, nullptr, {Range(101, 101)}));
}
TEST_P(BlobTableInteTest, TestDataEvolutionDvValidatesSelectedBlobRange) {
arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
BlobUtils::ToArrowField("b0", /*nullable=*/true)};
CreateTable(fields, {},
{{Options::FILE_FORMAT, GetParam()},
{Options::FILE_SYSTEM, "local"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
{Options::DELETION_VECTORS_ENABLED, "true"}});
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto normal = arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[0]}),
R"([[0],[1],[2],[3]])")
.ValueOrDie();
ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> base,
WriteArray(table_path, {}, {"f0"}, {normal}));
SetFirstRowId(100, base);
ASSERT_OK(Commit(table_path, base));
auto blobs = MakeBlobUpdateArray(fields[1], {"b0", "b1", "b2", "b3", "b4", "b5", "b6", "b7"});
ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> backfill,
WriteArray(table_path, {}, {"b0"}, {blobs}));
SetFirstRowId(98, backfill);
ASSERT_OK(Commit(table_path, backfill));
ASSERT_OK_AND_ASSIGN(std::string anchor,
DeletionVectorTestHelper::RetrieveAnchorFileName(base));
ASSERT_OK(CommitDeletionVectors(table_path, base[0], {{anchor, {1}}}));
auto expected = checked_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields),
R"([[0,"b2"],[2,"b4"],[3,"b5"]])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, {"f0", "b0"}, expected, nullptr, {Range(100, 103)}));
ASSERT_NOK_WITH_MSG(ScanAndRead(table_path, {"f0", "b0"}, nullptr),
"should contain selected row id range");
ASSERT_NOK_WITH_MSG(ScanAndRead(table_path, {"f0", "b0"}, expected, nullptr, {Range(99, 100)}),
"should contain selected row id range");
}
TEST_P(BlobTableInteTest, TestDataEvolutionRejectsAnchorlessGroupWithDeletionVectors) {
arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
BlobUtils::ToArrowField("b0", /*nullable=*/true)};
CreateTable(fields, {},
{{Options::FILE_FORMAT, GetParam()},
{Options::FILE_SYSTEM, "local"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
{Options::DELETION_VECTORS_ENABLED, "true"}});
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto normal =
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[0]}), R"([[0],[1],[2]])")
.ValueOrDie();
ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> base,
WriteArray(table_path, {}, {"f0"}, {normal}));
SetFirstRowId(100, base);
ASSERT_OK(Commit(table_path, base));
auto blobs = MakeBlobUpdateArray(fields[1], {"b0", "b1"});
ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> dedicated,
WriteArray(table_path, {}, {"b0"}, {blobs}));
SetFirstRowId(200, dedicated);
ASSERT_OK(Commit(table_path, dedicated));
ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> rows,
ScanAndReadResult(table_path, {"b0", "_ROW_ID"}));
ASSERT_TRUE(rows);
ASSERT_EQ(rows->length(), 5);
ASSERT_OK_AND_ASSIGN(std::string anchor,
DeletionVectorTestHelper::RetrieveAnchorFileName(base));
ASSERT_OK(CommitDeletionVectors(table_path, base[0], {{anchor, {1}}}));
// The same split now carries a DV. Match Java by requiring a normal anchor in each group.
ASSERT_NOK_WITH_MSG(ScanAndReadResult(table_path, {"b0", "_ROW_ID"}), "normal anchor file");
}
TEST_P(BlobTableInteTest, TestDataEvolutionBlobPartialUpdateCompactedLayers) {
arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
arrow::field("f1", arrow::utf8()), BlobUtils::ToArrowField("b0")};
std::map<std::string, std::string> options = {{Options::FILE_FORMAT, GetParam()},
{Options::FILE_SYSTEM, "local"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields);
// Mirrors the layer shapes of Java's BlobUpdateTest.testReadCompactedBlobSequenceGroups
// (with a single base write, as the C++ write path assigns one sequence per commit):
// row id: 0 1 2 3 4 5 6 7 8 9
// seq1: [b0 b1 b2 b3 b4 b5 b6 b7 b8 b9]
// seq2: [u20 P P u23 u24] . . . . .
// seq3: . . . . . [P u46 P u48 P]
// seq4: [P u61 P P P P P P P u69]
// result: u20 u61 b2 u23 u24 b5 u46 b7 u48 u69
auto base_array = PrepareBulkData(
10,
[](int32_t i) {
return std::to_string(i) + ", \"name_" + std::to_string(i) + "\", \"blob_" +
std::to_string(i) + "\"";
},
fields);
ASSERT_OK_AND_ASSIGN(auto commit_msgs0,
WriteArray(table_path, {}, schema->field_names(), {base_array}));
ASSERT_OK(Commit(table_path, commit_msgs0));
auto update_array1 = MakeBlobUpdateArray(fields[2], {"u20", "PH", "PH", "u23", "u24"});
ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, {}, {"b0"}, {update_array1}));
SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs1);
ASSERT_OK(Commit(table_path, commit_msgs1));
auto update_array2 = MakeBlobUpdateArray(fields[2], {"PH", "u46", "PH", "u48", "PH"});
ASSERT_OK_AND_ASSIGN(auto commit_msgs2, WriteArray(table_path, {}, {"b0"}, {update_array2}));
SetFirstRowId(/*reset_first_row_id=*/5, commit_msgs2);
ASSERT_OK(Commit(table_path, commit_msgs2));
auto update_array3 = MakeBlobUpdateArray(
fields[2], {"PH", "u61", "PH", "PH", "PH", "PH", "PH", "PH", "PH", "u69"});
ASSERT_OK_AND_ASSIGN(auto commit_msgs3, WriteArray(table_path, {}, {"b0"}, {update_array3}));
SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs3);
ASSERT_OK(Commit(table_path, commit_msgs3));
const std::vector<std::string> expected_blobs = {"u20", "u61", "blob_2", "u23", "u24",
"blob_5", "u46", "blob_7", "u48", "u69"};
auto expected_row = [&](int32_t i) {
return std::to_string(i) + ", \"name_" + std::to_string(i) + "\", \"" + expected_blobs[i] +
"\"";
};
auto expected_full = PrepareBulkData(10, expected_row, fields);
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_full));
// every row resolves independently under row-range pushdown
for (int32_t i = 0; i < 10; i++) {
auto expected_single = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields),
"[[" + expected_row(i) + "]]")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_single,
/*predicate=*/nullptr, /*row_ranges=*/{Range(i, i)}));
}
}
TEST_P(BlobTableInteTest, TestDataEvolutionBlobPartialUpdateWithRowRanges) {
arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
arrow::field("f1", arrow::utf8()), BlobUtils::ToArrowField("b0")};
std::map<std::string, std::string> options = {{Options::FILE_FORMAT, GetParam()},
{Options::FILE_SYSTEM, "local"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields);
auto base_array = PrepareBulkData(
10,
[](int32_t i) {
return std::to_string(i) + ", \"name_" + std::to_string(i) + "\", \"blob_" +
std::to_string(i) + "\"";
},
fields);
ASSERT_OK_AND_ASSIGN(auto commit_msgs0,
WriteArray(table_path, {}, schema->field_names(), {base_array}));
ASSERT_OK(Commit(table_path, commit_msgs0));
// partial update touching rows 1 and 9 only
auto update_array = MakeBlobUpdateArray(
fields[2], {"PH", "update_1", "PH", "PH", "PH", "PH", "PH", "PH", "PH", "update_9"});
ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, {}, {"b0"}, {update_array}));
SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs1);
ASSERT_OK(Commit(table_path, commit_msgs1));
// push down row ranges hitting updated and untouched rows
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
[1, "name_1", "update_1"],
[5, "name_5", "blob_5"],
[9, "name_9", "update_9"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array,
/*predicate=*/nullptr,
/*row_ranges=*/{Range(1, 1), Range(5, 5), Range(9, 9)}));
// full read still resolves every row
auto expected_full = PrepareBulkData(
10,
[](int32_t i) {
std::string blob = (i == 1 || i == 9) ? "\"update_" + std::to_string(i) + "\""
: "\"blob_" + std::to_string(i) + "\"";
return std::to_string(i) + ", \"name_" + std::to_string(i) + "\", " + blob;
},
fields);
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_full));
}
/// A BLOB sequence layer whose physical file covers only a strict subrange of the full row id
/// range: the newest group is internally Gap(2), File([u2, u3]), Gap(6). The table needs a
/// normal column to be creatable, but only the blob column is ever written, so the split
/// contains only blob files and the blob bunch itself carries the row-tracking fields: the
/// fallback reader must keep _ROW_ID correct and report each row's _SEQUENCE_NUMBER from the
/// layer that resolved it, without ever exposing the internal placeholder sentinel.
TEST_P(BlobTableInteTest, TestDataEvolutionBlobPartialUpdateRowTrackingWithSubrangeLayer) {
arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
BlobUtils::ToArrowField("b0", /*nullable=*/true)};
std::map<std::string, std::string> options = {{Options::FILE_FORMAT, GetParam()},
{Options::FILE_SYSTEM, "local"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
// old layer: rows 0-9 all real
std::vector<std::optional<std::string>> base_rows;
for (int32_t i = 0; i < 10; i++) {
base_rows.emplace_back("b" + std::to_string(i));
}
auto base_array = MakeBlobUpdateArray(fields[1], base_rows);
ASSERT_OK_AND_ASSIGN(auto commit_msgs0, WriteArray(table_path, {}, {"b0"}, {base_array}));
SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs0);
ASSERT_OK(Commit(table_path, commit_msgs0));
// new layer: first_row_id=2, row_count=2, covering only rows 2-3
auto update_array = MakeBlobUpdateArray(fields[1], {"u2", "u3"});
ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, {}, {"b0"}, {update_array}));
SetFirstRowId(/*reset_first_row_id=*/2, commit_msgs1);
ASSERT_OK(Commit(table_path, commit_msgs1));
ASSERT_OK_AND_ASSIGN(auto scan_read,
ScanAndReadResult(table_path, {"b0", "_ROW_ID", "_SEQUENCE_NUMBER"}));
ASSERT_TRUE(scan_read);
auto concat_array = arrow::Concatenate(scan_read->chunks()).ValueOrDie();
auto struct_array = std::dynamic_pointer_cast<arrow::StructArray>(concat_array);
ASSERT_TRUE(struct_array);
ASSERT_EQ(struct_array->length(), 10);
auto blob_col =
std::dynamic_pointer_cast<arrow::LargeBinaryArray>(struct_array->GetFieldByName("b0"));
auto row_id_col = std::dynamic_pointer_cast<arrow::Int64Array>(
struct_array->GetFieldByName(SpecialFields::RowId().Name()));
auto seq_col = std::dynamic_pointer_cast<arrow::Int64Array>(
struct_array->GetFieldByName(SpecialFields::SequenceNumber().Name()));
ASSERT_TRUE(blob_col && row_id_col && seq_col);
int64_t old_layer_seq = seq_col->Value(0);
int64_t new_layer_seq = seq_col->Value(2);
ASSERT_GT(new_layer_seq, old_layer_seq);
for (int64_t i = 0; i < 10; i++) {
ASSERT_FALSE(blob_col->IsNull(i));
std::string expected_blob =
(i == 2 || i == 3) ? "u" + std::to_string(i) : "b" + std::to_string(i);
// the leading and trailing gaps never expose the internal placeholder sentinel
ASSERT_EQ(blob_col->GetString(i), expected_blob) << "row " << i;
ASSERT_EQ(row_id_col->Value(i), i);
ASSERT_EQ(seq_col->Value(i), (i == 2 || i == 3) ? new_layer_seq : old_layer_seq)
<< "row " << i;
}
// row-range pushdown with the row-tracking projection: the selection removes rows inside
// the blob files, so batch positions must still map back to the right file row indexes
ASSERT_OK_AND_ASSIGN(auto range_read,
ScanAndReadResult(table_path, {"b0", "_ROW_ID", "_SEQUENCE_NUMBER"},
/*predicate=*/nullptr,
/*row_ranges=*/{Range(1, 2), Range(8, 8)}));
ASSERT_TRUE(range_read);
auto range_concat = arrow::Concatenate(range_read->chunks()).ValueOrDie();
auto range_struct = std::dynamic_pointer_cast<arrow::StructArray>(range_concat);
ASSERT_TRUE(range_struct);
ASSERT_EQ(range_struct->length(), 3);
auto range_blob_col =
std::dynamic_pointer_cast<arrow::LargeBinaryArray>(range_struct->GetFieldByName("b0"));
auto range_row_id_col = std::dynamic_pointer_cast<arrow::Int64Array>(
range_struct->GetFieldByName(SpecialFields::RowId().Name()));
auto range_seq_col = std::dynamic_pointer_cast<arrow::Int64Array>(
range_struct->GetFieldByName(SpecialFields::SequenceNumber().Name()));
ASSERT_TRUE(range_blob_col && range_row_id_col && range_seq_col);
const std::vector<int64_t> expected_row_ids = {1, 2, 8};
const std::vector<std::string> expected_blobs = {"b1", "u2", "b8"};
for (int64_t i = 0; i < 3; i++) {
ASSERT_EQ(range_blob_col->GetString(i), expected_blobs[i]) << "row " << i;
ASSERT_EQ(range_row_id_col->Value(i), expected_row_ids[i]) << "row " << i;
ASSERT_EQ(range_seq_col->Value(i), expected_row_ids[i] == 2 ? new_layer_seq : old_layer_seq)
<< "row " << i;
}
}
/// A row that is a placeholder in every BLOB layer degrades to a null blob but keeps its
/// _ROW_ID and reports -1 as its _SEQUENCE_NUMBER. Only the blob column is ever written, so
/// the blob bunch itself carries the row-tracking fields.
TEST_P(BlobTableInteTest, TestDataEvolutionBlobPartialUpdateAllPlaceholderRowTracking) {
arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
BlobUtils::ToArrowField("b0", /*nullable=*/true)};
std::map<std::string, std::string> options = {{Options::FILE_FORMAT, GetParam()},
{Options::FILE_SYSTEM, "local"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
// the base layer itself holds a placeholder at row 1, so after the second layer also
// leaves it untouched, row 1 is a placeholder in every layer
auto base_array = MakeBlobUpdateArray(fields[1], {"blob_a", "PH", "blob_c"});
ASSERT_OK_AND_ASSIGN(auto commit_msgs0, WriteArray(table_path, {}, {"b0"}, {base_array}));
SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs0);
ASSERT_OK(Commit(table_path, commit_msgs0));
auto update_array = MakeBlobUpdateArray(fields[1], {"PH", "PH", "update_c"});
ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, {}, {"b0"}, {update_array}));
SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs1);
ASSERT_OK(Commit(table_path, commit_msgs1));
ASSERT_OK_AND_ASSIGN(auto scan_read,
ScanAndReadResult(table_path, {"b0", "_ROW_ID", "_SEQUENCE_NUMBER"}));
ASSERT_TRUE(scan_read);
auto concat_array = arrow::Concatenate(scan_read->chunks()).ValueOrDie();
auto struct_array = std::dynamic_pointer_cast<arrow::StructArray>(concat_array);
ASSERT_TRUE(struct_array);
ASSERT_EQ(struct_array->length(), 3);
auto blob_col =
std::dynamic_pointer_cast<arrow::LargeBinaryArray>(struct_array->GetFieldByName("b0"));
auto row_id_col = std::dynamic_pointer_cast<arrow::Int64Array>(
struct_array->GetFieldByName(SpecialFields::RowId().Name()));
auto seq_col = std::dynamic_pointer_cast<arrow::Int64Array>(
struct_array->GetFieldByName(SpecialFields::SequenceNumber().Name()));
ASSERT_TRUE(blob_col && row_id_col && seq_col);
ASSERT_EQ(blob_col->GetString(0), "blob_a");
ASSERT_EQ(blob_col->GetString(2), "update_c");
// the all-placeholder row: null blob, row id kept, sequence number -1
ASSERT_TRUE(blob_col->IsNull(1));
ASSERT_EQ(row_id_col->Value(1), 1);
ASSERT_EQ(seq_col->Value(1), -1);
for (int64_t i : {static_cast<int64_t>(0), static_cast<int64_t>(2)}) {
ASSERT_EQ(row_id_col->Value(i), i);
ASSERT_GE(seq_col->Value(i), 0);
}
ASSERT_GT(seq_col->Value(2), seq_col->Value(0));
}
/// Placeholders are identified by exact equality with the sentinel bytes (see
/// BlobDefs::kPlaceholderSentinel): a blob value that merely starts with them is a normal value,
/// stored verbatim by a full-row write and by a partial update alike.
TEST_P(BlobTableInteTest, TestBlobValuePrefixedWithPlaceholderSentinelBytes) {
arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
arrow::field("f1", arrow::utf8()),
BlobUtils::ToArrowField("b0", /*nullable=*/true)};
std::map<std::string, std::string> options = {{Options::FILE_FORMAT, GetParam()},
{Options::FILE_SYSTEM, "local"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields);
std::string sentinel_bytes = std::string(BlobDefs::PlaceholderSentinelView());
std::string prefixed_bytes = sentinel_bytes + "suffix";
// full-row write: row 0's blob merely starts with the sentinel bytes
std::string src_json =
std::string(R"([[1, "a", ")") + prefixed_bytes + R"("], [2, "b", "normal"]])";
auto src_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), src_json).ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs0,
WriteArray(table_path, {}, schema->field_names(), {src_array}));
ASSERT_OK(Commit(table_path, commit_msgs0));
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), src_array));
// partial update: row 0 falls back to the prefixed value, row 1 is updated to a value that
// is the sentinel bytes twice
std::string doubled_bytes = sentinel_bytes + sentinel_bytes;
auto update_array = MakeBlobUpdateArray(fields[2], {"PH", doubled_bytes});
ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, {}, {"b0"}, {update_array}));
SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs1);
ASSERT_OK(Commit(table_path, commit_msgs1));
std::string expected_json = std::string(R"([[1, "a", ")") + prefixed_bytes +
R"("], [2, "b", ")" + doubled_bytes + R"("]])";
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), expected_json)
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
}
TEST_P(BlobTableInteTest, TestBlobSentinelValueInBaseLayerDegradesToNull) {
// Pins the accepted collision of the byte-identified placeholder protocol (see
// BlobDefs::kPlaceholderSentinel): every write persists a user blob equal to the sentinel
// bytes as a placeholder entry, so once a later partial update leaves the row untouched it
// is a placeholder in every layer and degrades to a null blob instead of falling back.
arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
arrow::field("f1", arrow::utf8()),
BlobUtils::ToArrowField("b0", /*nullable=*/true)};
std::map<std::string, std::string> options = {{Options::FILE_FORMAT, GetParam()},
{Options::FILE_SYSTEM, "local"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields);
std::string sentinel_bytes = std::string(BlobDefs::PlaceholderSentinelView());
// the full-row write persists row 0's sentinel-equal bytes as a placeholder entry
std::string src_json =
std::string(R"([[1, "a", ")") + sentinel_bytes + R"("], [2, "b", "blob_b"]])";
auto src_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), src_json).ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs0,
WriteArray(table_path, {}, schema->field_names(), {src_array}));
ASSERT_OK(Commit(table_path, commit_msgs0));
// the partial update leaves row 0 untouched, so its base value collides with the
// placeholder markers of the newer layer
auto update_array = MakeBlobUpdateArray(fields[2], {"PH", "updated_b"});
ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, {}, {"b0"}, {update_array}));
SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs1);
ASSERT_OK(Commit(table_path, commit_msgs1));
// row 0's non-blob fields survive (they come from the data file); only the blob degrades
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
[1, "a", null],
[2, "b", "updated_b"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
}
TEST_P(BlobTableInteTest, TestSingleBlobLayerPlaceholderBecomesNull) {
// blob.internal.emit-placeholder-sentinel is reserved for the data-evolution fallback read
// path. A single layer also uses fallback: an unresolved placeholder becomes null, and the
// internal sentinel must never be exposed as the blob value.
arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
arrow::field("f1", arrow::utf8()),
BlobUtils::ToArrowField("b0", /*nullable=*/true)};
std::map<std::string, std::string> options = {{Options::FILE_FORMAT, GetParam()},
{Options::FILE_SYSTEM, "local"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
{BlobDefs::kEmitPlaceholderSentinelKey, "true"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields);
std::string sentinel_bytes = std::string(BlobDefs::PlaceholderSentinelView());
std::string src_json =
std::string(R"([[1, "a", ")") + sentinel_bytes + R"("], [2, "b", "blob_b"]])";
auto src_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), src_json).ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs,
WriteArray(table_path, {}, schema->field_names(), {src_array}));
ASSERT_OK(Commit(table_path, commit_msgs));
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields),
R"([[1, "a", null], [2, "b", "blob_b"]])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
}
TEST_P(BlobTableInteTest, TestDataEvolutionBlobOnlyFirstCommitFailsWithoutFirstRowId) {
arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
arrow::field("f1", arrow::utf8()), BlobUtils::ToArrowField("b0")};
std::map<std::string, std::string> options = {{Options::FILE_FORMAT, GetParam()},
{Options::FILE_SYSTEM, "local"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto blob_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[2]}), R"([
["blob_0"]
])")
.ValueOrDie());
std::vector<std::string> blob_write_cols = {"b0"};
ASSERT_OK_AND_ASSIGN(auto commit_msgs,
WriteArray(table_path, {}, blob_write_cols, {blob_array}));
ASSERT_EQ(commit_msgs.size(), 1);
auto commit_msg = std::dynamic_pointer_cast<CommitMessageImpl>(commit_msgs[0]);
ASSERT_TRUE(commit_msg);
ASSERT_EQ(commit_msg->data_increment_.new_files_.size(), 1);
const auto& blob_file = commit_msg->data_increment_.new_files_[0];
ASSERT_TRUE(BlobUtils::IsBlobFile(blob_file->file_name));
ASSERT_FALSE(blob_file->first_row_id.has_value());
ASSERT_NOK_WITH_MSG(Commit(table_path, commit_msgs), "blobStart 0 should be less than start 0");
}
TEST_P(BlobTableInteTest, TestMultipleAppendsDifferentFirstRowIds) {
CreateTable();
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields_);
// First commit, firstRowId = 0
// write field: f0, f1
std::vector<std::string> write_cols1 = {"f0", "f1"};
auto src_array1 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[0], fields_[1]}), R"([
[1, "a"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, {}, write_cols1, {src_array1}));
SetFirstRowId(0, commit_msgs1);
// write field: f2
std::vector<std::string> write_cols2 = {"f2"};
auto src_array2 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[2]}), R"([
["b"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs2, WriteArray(table_path, {}, write_cols2, {src_array2}));
SetFirstRowId(0, commit_msgs2);
std::vector<std::shared_ptr<CommitMessage>> total_msgs;
total_msgs.insert(total_msgs.end(), commit_msgs1.begin(), commit_msgs1.end());
total_msgs.insert(total_msgs.end(), commit_msgs2.begin(), commit_msgs2.end());
ASSERT_OK(Commit(table_path, total_msgs));
// Second commit, firstRowId = 1
// write field: f0, f1
std::vector<std::string> write_cols3 = {"f0", "f1"};
auto src_array3 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[0], fields_[1]}), R"([
[2, "c"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs3, WriteArray(table_path, {}, write_cols3, {src_array3}));
SetFirstRowId(1, commit_msgs3);
ASSERT_OK(Commit(table_path, commit_msgs3));
// Third commit
// write field: f2
std::vector<std::string> write_cols4 = {"f2"};
auto src_array4 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[2]}), R"([
["d"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs4, WriteArray(table_path, {}, write_cols4, {src_array4}));
SetFirstRowId(1, commit_msgs4);
ASSERT_OK(Commit(table_path, commit_msgs4));
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
[1, "a", "b"],
[2, "c", "d"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
// read with row tracking
auto expected_row_tracking_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({
fields_[0],
fields_[1],
fields_[2],
SpecialFields::RowId().ArrowField(),
SpecialFields::SequenceNumber().ArrowField(),
}),
R"([
[1, "a", "b", 0, 1],
[2, "c", "d", 1, 3]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, {"f0", "f1", "f2", "_ROW_ID", "_SEQUENCE_NUMBER"},
expected_row_tracking_array));
}
TEST_P(BlobTableInteTest, TestMoreDataWithDataEvolution) {
CreateTable();
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields_);
// write field: f0, f1
std::vector<std::string> write_cols1 = {"f0", "f1"};
// row0: 0, a0; row1: 1, a1 ...
auto src_array1 = PrepareBulkData(
10000, [](int32_t i) { return std::to_string(i) + ", \"a" + std::to_string(i) + "\""; },
{fields_[0], fields_[1]});
ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, {}, write_cols1, {src_array1}));
SetFirstRowId(0, commit_msgs1);
// write field: f2
std::vector<std::string> write_cols2 = {"f2"};
// row0: b0; row1: b1 ...
auto src_array2 = PrepareBulkData(
10000, [](int32_t i) { return "\"b" + std::to_string(i) + "\""; }, {fields_[2]});
ASSERT_OK_AND_ASSIGN(auto commit_msgs2, WriteArray(table_path, {}, write_cols2, {src_array2}));
SetFirstRowId(0, commit_msgs2);
std::vector<std::shared_ptr<CommitMessage>> total_msgs;
total_msgs.insert(total_msgs.end(), commit_msgs1.begin(), commit_msgs1.end());
total_msgs.insert(total_msgs.end(), commit_msgs2.begin(), commit_msgs2.end());
ASSERT_OK(Commit(table_path, total_msgs));
// write field: f2
std::vector<std::string> write_cols3 = {"f2"};
// row0: c0; row1: c1 ...
auto src_array3 = PrepareBulkData(
10000, [](int32_t i) { return "\"c" + std::to_string(i) + "\""; }, {fields_[2]});
ASSERT_OK_AND_ASSIGN(auto commit_msgs3, WriteArray(table_path, {}, write_cols3, {src_array3}));
ASSERT_OK(Commit(table_path, commit_msgs3));
// row0: 0, a0, c0; row1: 1, a1, c1 ...
auto expected_array = PrepareBulkData(10000,
[](int32_t i) {
return std::to_string(i) + ", \"a" +
std::to_string(i) + "\", \"c" +
std::to_string(i) + "\"";
},
{fields_[0], fields_[1], fields_[2]});
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
}
TEST_P(BlobTableInteTest, TestBlobWriteMultiRound) {
std::map<std::string, std::string> options = {
{Options::FILE_FORMAT, GetParam()}, {Options::FILE_SYSTEM, "local"},
{Options::ROW_TRACKING_ENABLED, "true"}, {Options::BLOB_TARGET_FILE_SIZE, "1000"},
{Options::TARGET_FILE_SIZE, "100"}, {Options::DATA_EVOLUTION_ENABLED, "true"}};
CreateTable(/*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields_);
// write field: f0, f1, f2
std::vector<std::string> write_cols = schema->field_names();
int32_t offset = 0;
auto array_1 = PrepareBulkData(5000,
[offset](int32_t i) {
return std::to_string(offset + i) + ", \"a" +
std::to_string(offset + i) + "\", \"c" +
std::to_string(offset + i) + "\"";
},
{fields_[0], fields_[1], fields_[2]});
offset = 5000;
auto array_2 = PrepareBulkData(5000,
[offset](int32_t i) {
return std::to_string(offset + i) + ", \"a" +
std::to_string(offset + i) + "\", \"c" +
std::to_string(offset + i) + "\"";
},
{fields_[0], fields_[1], fields_[2]});
ASSERT_OK_AND_ASSIGN(auto commit_msgs,
WriteArray(table_path, {}, write_cols, {array_1, array_2}));
ASSERT_OK(Commit(table_path, commit_msgs));
auto concat_array = arrow::Concatenate({array_1, array_2}).ValueOrDie();
auto expect_array = std::dynamic_pointer_cast<arrow::StructArray>(concat_array);
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expect_array));
}
TEST_P(BlobTableInteTest, TestExternalPath) {
// create external path dir
auto external_dir = UniqueTestDirectory::Create("local");
ASSERT_TRUE(external_dir);
std::string external_test_dir = external_dir->Str();
std::map<std::string, std::string> options = {
{Options::FILE_FORMAT, GetParam()},
{Options::FILE_SYSTEM, "local"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
{Options::DATA_FILE_EXTERNAL_PATHS, "FILE://" + external_test_dir}};
CreateTable(/*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields_);
// write field: f0, f1
std::vector<std::string> write_cols0 = {"f0", "f1"};
auto src_array0 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[0], fields_[1]}), R"([
[1, "a"],
[2, "c"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs0, WriteArray(table_path, {}, write_cols0, {src_array0}));
ASSERT_OK(Commit(table_path, commit_msgs0));
// write field: f0, f2
std::vector<std::string> write_cols1 = {"f0", "f2"};
auto src_array1 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[0], fields_[2]}), R"([
[10, "b"],
[20, "d"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, {}, write_cols1, {src_array1}));
SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs1);
ASSERT_OK(Commit(table_path, commit_msgs1));
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
[10, "a", "b"],
[20, "c", "d"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
// read with row tracking
auto expected_row_tracking_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(
arrow::struct_({fields_[1], fields_[0], fields_[2], SpecialFields::RowId().ArrowField(),
SpecialFields::SequenceNumber().ArrowField()}),
R"([
["a", 10, "b", 0, 2],
["c", 20, "d", 1, 2]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, {"f1", "f0", "f2", "_ROW_ID", "_SEQUENCE_NUMBER"},
expected_row_tracking_array));
}
TEST_P(BlobTableInteTest, TestPartitionWithPredicate) {
auto file_format = GetParam();
std::vector<std::string> partition_keys = {"f0"};
std::map<std::string, std::string> options = {{Options::FILE_FORMAT, GetParam()},
{Options::FILE_SYSTEM, "local"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
{"parquet.write.max-row-group-length", "1"}};
CreateTable(partition_keys, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields_);
// write field: f0, f1 for partition f0 = "11"
std::vector<std::string> write_cols0 = {"f0", "f1"};
auto src_array0 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[0], fields_[1]}), R"([
[11, "2024"],
[11, "2025"],
[11, "2026"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs0,
WriteArray(table_path, {{"f0", "11"}}, write_cols0, {src_array0}));
// write field: f2 for partition f0 = "11"
std::vector<std::string> write_cols1 = {"f2"};
auto src_array1 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[2]}), R"([
["a"],
["b"],
["c"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs1,
WriteArray(table_path, {{"f0", "11"}}, write_cols1, {src_array1}));
std::vector<std::shared_ptr<CommitMessage>> total_msgs;
total_msgs.insert(total_msgs.end(), commit_msgs0.begin(), commit_msgs0.end());
total_msgs.insert(total_msgs.end(), commit_msgs1.begin(), commit_msgs1.end());
SetFirstRowId(0, total_msgs);
ASSERT_OK(Commit(table_path, total_msgs));
// write field: f0, f1 for partition f0 = "22"
std::vector<std::string> write_cols3 = {"f0", "f1"};
auto src_array3 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[0], fields_[1]}), R"([
[22, "2027"],
[22, "2028"],
[22, "2029"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs3,
WriteArray(table_path, {{"f0", "22"}}, write_cols3, {src_array3}));
SetFirstRowId(3, commit_msgs3);
ASSERT_OK(Commit(table_path, commit_msgs3));
{
// only set data field predicate, file with field all null will be returned
auto equal = PredicateBuilder::Equal(/*field_index=*/2, /*field_name=*/"f2",
FieldType::STRING, Literal(FieldType::STRING, "a", 1));
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
[11, "2024", "a"],
[11, "2025", "b"],
[11, "2026", "c"],
[22, "2027", null],
[22, "2028", null],
[22, "2029", null]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array, equal));
}
{
// only set partition predicate
auto equal = PredicateBuilder::Equal(/*field_index=*/0, /*field_name=*/"f0", FieldType::INT,
Literal(11));
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
[11, "2024", "a"],
[11, "2025", "b"],
[11, "2026", "c"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array, equal));
}
{
// set partition predicate and data field predicate, blob type not support predicate
auto equal = PredicateBuilder::Equal(/*field_index=*/1, /*field_name=*/"f1",
FieldType::BLOB, Literal(FieldType::BLOB, "2024", 4));
auto greater_than = PredicateBuilder::GreaterThan(/*field_index=*/0, /*field_name=*/"f0",
FieldType::INT, Literal(100));
ASSERT_OK_AND_ASSIGN(auto predicate, PredicateBuilder::And({equal, greater_than}));
ASSERT_NOK_WITH_MSG(
ScanAndRead(table_path, schema->field_names(), /*expected_array=*/nullptr, predicate),
"Invalid type large_binary for predicate");
}
{
// read with row tracking
auto equal = PredicateBuilder::Equal(/*field_index=*/0, /*field_name=*/"f0", FieldType::INT,
Literal(11));
auto expected_row_tracking_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(
arrow::struct_({fields_[0], fields_[1], fields_[2],
SpecialFields::RowId().ArrowField(),
SpecialFields::SequenceNumber().ArrowField()}),
R"([
[11, "2024", "a", 0, 1],
[11, "2025", "b", 1, 1],
[11, "2026", "c", 2, 1]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, {"f0", "f1", "f2", "_ROW_ID", "_SEQUENCE_NUMBER"},
expected_row_tracking_array, equal));
}
}
TEST_P(BlobTableInteTest, TestPredicate) {
if (GetParam() == "avro") {
// Avro does not have stats.
return;
}
if (GetParam() == "mosaic") {
CreateTable(/*partition_keys=*/{}, {{Options::FILE_FORMAT, GetParam()},
{Options::FILE_SYSTEM, "local"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
{mosaic::MOSAIC_STATS_COLUMNS, "f0,f2"}});
} else {
CreateTable();
}
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields_);
// write field: f0, f1, f2
std::vector<std::string> write_cols0 = schema->field_names();
auto src_array0 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
[1, "a", "b"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs, WriteArray(table_path, {}, write_cols0, {src_array0}));
ASSERT_OK(Commit(table_path, commit_msgs));
// write field: f2
std::vector<std::string> write_cols1 = {"f2"};
auto src_array1 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[2]}), R"([
["c"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(commit_msgs, WriteArray(table_path, {}, write_cols1, {src_array1}));
SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs);
ASSERT_OK(Commit(table_path, commit_msgs));
{
// test no predicate
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
[1, "a", "c"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
}
{
// test predicate with f2
auto predicate =
PredicateBuilder::NotEqual(/*field_index=*/2, /*field_name=*/"f2", FieldType::STRING,
Literal(FieldType::STRING, "b", 1));
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
[1, "a", "c"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array, predicate));
}
{
// test predicate with f1
auto predicate =
PredicateBuilder::NotEqual(/*field_index=*/1, /*field_name=*/"f1", FieldType::STRING,
Literal(FieldType::STRING, "a", 1));
ASSERT_NOK_WITH_MSG(
ScanAndRead(table_path, schema->field_names(), /*expected_array=*/nullptr, predicate),
"Invalid type large_binary for predicate");
}
{
// test predicate with f2
auto predicate =
PredicateBuilder::NotEqual(/*field_index=*/2, /*field_name=*/"f2", FieldType::STRING,
Literal(FieldType::STRING, "c", 1));
ASSERT_OK(
ScanAndRead(table_path, schema->field_names(), /*expected_array=*/nullptr, predicate));
}
}
TEST_P(BlobTableInteTest, TestIOException) {
std::string table_path;
// write and commit with I/O exception
bool write_run_complete = false;
auto io_hook = IOHook::GetInstance();
for (size_t i = 0; i < 2000; i += paimon::test::RandomNumber(20, 30)) {
ScopeGuard guard([&io_hook]() { io_hook->Clear(); });
dir_ = UniqueTestDirectory::Create("local");
CreateTable();
table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields_);
io_hook->Reset(i, IOHook::Mode::RETURN_ERROR);
// write field: f0, f1, f2
std::vector<std::string> write_cols0 = schema->field_names();
auto src_array0 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
[10, "a", "b"],
[20, "aa", "bb"],
[23, "aaa", "bbb"]
])")
.ValueOrDie());
auto commit_msgs0_result = WriteArray(table_path, {}, write_cols0, {src_array0});
CHECK_HOOK_STATUS(commit_msgs0_result.status(), i);
CHECK_HOOK_STATUS(Commit(table_path, commit_msgs0_result.value()), i);
// write field: f2, f0
std::vector<std::string> write_cols1 = {"f2", "f0"};
auto src_array1 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[2], fields_[0]}), R"([
["c", 100],
["cc", 200],
["ccc", 300]
])")
.ValueOrDie());
auto commit_msgs1_result = WriteArray(table_path, {}, write_cols1, {src_array1});
CHECK_HOOK_STATUS(commit_msgs1_result.status(), i);
auto commit_msgs1 = std::move(commit_msgs1_result).value();
SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs1);
CHECK_HOOK_STATUS(Commit(table_path, commit_msgs1), i);
write_run_complete = true;
break;
}
ASSERT_TRUE(write_run_complete);
// scan and read with I / O exception
bool read_run_complete = false;
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(
arrow::struct_({fields_[1], fields_[0], SpecialFields::SequenceNumber().ArrowField(),
SpecialFields::RowId().ArrowField(), fields_[2]}),
R"([
["a", 100, 2, 0, "c"],
["aa", 200, 2, 1, "cc"],
["aaa", 300, 2, 2, "ccc"]
])")
.ValueOrDie());
for (size_t i = 0; i < 2000; i++) {
ScopeGuard guard([&io_hook]() { io_hook->Clear(); });
io_hook->Reset(i, IOHook::Mode::RETURN_ERROR);
CHECK_HOOK_STATUS(ScanAndRead(table_path, {"f1", "f0", "_SEQUENCE_NUMBER", "_ROW_ID", "f2"},
expected_array),
i);
read_run_complete = true;
break;
}
ASSERT_TRUE(read_run_complete);
}
TEST_P(BlobTableInteTest, TestReadTableWithDenseStats) {
auto file_format = GetParam();
if (file_format == "avro" || file_format == "mosaic") {
return;
}
std::string table_path =
paimon::test::GetDataDir() + file_format +
"/blob_data_evolution_with_dense_stats.db/blob_data_evolution_with_dense_stats/";
std::vector<DataField> read_fields = {DataField(0, BlobUtils::ToArrowField("f0")),
DataField(1, arrow::field("f1", arrow::int32())),
DataField(2, arrow::field("f2", arrow::int32())),
DataField(3, arrow::field("f3", arrow::float64())),
SpecialFields::RowId(),
SpecialFields::SequenceNumber()};
std::shared_ptr<arrow::DataType> arrow_data_type =
DataField::ConvertDataFieldsToArrowStructType(read_fields);
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type, R"([
["Lily", 2, 102, 2.1, 0, 2],
["Alice", 4, 104, 3.1, 1, 2]
])")
.ValueOrDie());
{
ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
expected_array));
}
{
auto predicate =
PredicateBuilder::Equal(/*field_index=*/0, /*field_name=*/"f0", FieldType::BLOB,
Literal(FieldType::BLOB, "Alice", 5));
ASSERT_NOK_WITH_MSG(
ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
expected_array, predicate),
"Invalid type large_binary for predicate");
}
{
auto predicate = PredicateBuilder::Equal(/*field_index=*/2, /*field_name=*/"f2",
FieldType::INT, Literal(102));
ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
expected_array, predicate));
}
{
auto predicate = PredicateBuilder::Equal(/*field_index=*/2, /*field_name=*/"f2",
FieldType::INT, Literal(12));
ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
/*expected_array=*/nullptr, predicate));
}
{
auto predicate = PredicateBuilder::GreaterThan(/*field_index=*/1, /*field_name=*/"f1",
FieldType::INT, Literal(6));
ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
/*expected_array=*/nullptr, predicate));
}
{
// f3 does not have stats, therefore data will not be filtered
auto predicate = PredicateBuilder::GreaterThan(/*field_index=*/3, /*field_name=*/"f3",
FieldType::DOUBLE, Literal(5.1));
ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
expected_array, predicate));
}
}
TEST_P(BlobTableInteTest, TestDataEvolutionAndAlterTable) {
auto file_format = GetParam();
if (file_format == "avro" || file_format == "mosaic") {
return;
}
std::string table_path = paimon::test::GetDataDir() + file_format +
"/blob_append_table_alter_table_with_cast_with_data_evolution.db/"
"blob_append_table_alter_table_with_cast_with_data_evolution/";
std::vector<DataField> read_fields = {
DataField(6, arrow::field("f4", arrow::timestamp(arrow::TimeUnit::NANO))),
DataField(0, arrow::field("key0", arrow::int32())),
DataField(1, arrow::field("key1", arrow::int32())),
DataField(2, arrow::field("f3", arrow::int32())),
DataField(3, arrow::field("f1", arrow::utf8())),
DataField(4, arrow::field("f2", arrow::decimal128(6, 3))),
DataField(5, arrow::field("f0", arrow::boolean())),
DataField(8, BlobUtils::ToArrowField("blob")),
DataField(9, arrow::field("f6", arrow::int32())),
SpecialFields::RowId(),
SpecialFields::SequenceNumber()};
std::shared_ptr<arrow::DataType> arrow_data_type =
DataField::ConvertDataFieldsToArrowStructType(read_fields);
{
// only read blob column
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(
arrow::struct_({BlobUtils::ToArrowField("blob")}), R"([
["Lily"],
["Alice"],
["Bob"],
["Cindy"],
["Dave"],
["Apple"],
["Banana"],
["Cherry"],
["Durian"],
["Elderberry"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, {"blob"}, expected_array));
}
{
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type, R"([
["1970-01-05T00:00", 0, 1, 100, "2024-11-26 06:38:56.001000001", "0.020", true, "Lily", null, 0, 1],
["1969-11-18T00:00", 0, 1, 110, "2024-11-26 06:38:56.011000011", "11.120", true, "Alice", null, 1, 1],
["1971-03-21T00:00", 0, 1, 120, "2024-11-26 06:38:56.021000021", "22.220", false, "Bob", null, 2, 1],
["1957-11-01T00:00", 0, 1, 130, "2024-11-26 06:38:56.031000031", "333.320", false, "Cindy", null, 3, 1],
["2091-09-07T00:00", 0, 1, 140, "2024-11-26 06:38:56.041000041", "444.420", true, "Dave", null, 4, 1],
["2024-11-26T06:38:56.054000154", 0, 1, 150, "2024-11-26 15:28:31", "55.002", true, "Apple", 56, 5, 3],
["2024-11-26T06:38:56.064000164", 0, 1, 160, "2024-11-26 15:28:41", "666.012", false, "Banana", 66, 6, 3],
["2024-11-26T06:38:56.074000174", 0, 1, 170, "2024-11-26 15:28:51", "-77.022", true, "Cherry", 76, 7, 3],
["2024-11-26T06:38:56.084000184", 0, 1, 180, "2024-11-26 15:29:01", "8.032", true, "Durian", -86, 8, 3],
["2024-11-26T06:38:56.094000194", 0, 1, 190, "I'm strange", "-999.420", false, "Elderberry", 96, 9, 3]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
expected_array));
}
{
// only files with schema-1 will be skipped, while files with schema-0 will be reserved
// as type change
auto predicate = PredicateBuilder::GreaterThan(/*field_index=*/3, /*field_name=*/"f3",
FieldType::INT, Literal(200));
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type, R"([
["1970-01-05T00:00", 0, 1, 100, "2024-11-26 06:38:56.001000001", "0.020", true, "Lily", null, 0, 1],
["1969-11-18T00:00", 0, 1, 110, "2024-11-26 06:38:56.011000011", "11.120", true, "Alice", null, 1, 1],
["1971-03-21T00:00", 0, 1, 120, "2024-11-26 06:38:56.021000021", "22.220", false, "Bob", null, 2, 1],
["1957-11-01T00:00", 0, 1, 130, "2024-11-26 06:38:56.031000031", "333.320", false, "Cindy", null, 3, 1],
["2091-09-07T00:00", 0, 1, 140, "2024-11-26 06:38:56.041000041", "444.420", true, "Dave", null, 4, 1]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
expected_array, predicate));
}
{
// files with schema-0 will be reserved as f6 does not exist in schema-0 (null count =
// null), files with schema-1 will also be reserved
auto predicate =
PredicateBuilder::IsNotNull(/*field_index=*/8, /*field_name=*/"f6", FieldType::INT);
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type, R"([
["1970-01-05T00:00", 0, 1, 100, "2024-11-26 06:38:56.001000001", "0.020", true, "Lily", null, 0, 1],
["1969-11-18T00:00", 0, 1, 110, "2024-11-26 06:38:56.011000011", "11.120", true, "Alice", null, 1, 1],
["1971-03-21T00:00", 0, 1, 120, "2024-11-26 06:38:56.021000021", "22.220", false, "Bob", null, 2, 1],
["1957-11-01T00:00", 0, 1, 130, "2024-11-26 06:38:56.031000031", "333.320", false, "Cindy", null, 3, 1],
["2091-09-07T00:00", 0, 1, 140, "2024-11-26 06:38:56.041000041", "444.420", true, "Dave", null, 4, 1],
["2024-11-26T06:38:56.054000154", 0, 1, 150, "2024-11-26 15:28:31", "55.002", true, "Apple", 56, 5, 3],
["2024-11-26T06:38:56.064000164", 0, 1, 160, "2024-11-26 15:28:41", "666.012", false, "Banana", 66, 6, 3],
["2024-11-26T06:38:56.074000174", 0, 1, 170, "2024-11-26 15:28:51", "-77.022", true, "Cherry", 76, 7, 3],
["2024-11-26T06:38:56.084000184", 0, 1, 180, "2024-11-26 15:29:01", "8.032", true, "Durian", -86, 8, 3],
["2024-11-26T06:38:56.094000194", 0, 1, 190, "I'm strange", "-999.420", false, "Elderberry", 96, 9, 3]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
expected_array, predicate));
}
}
TEST_P(BlobTableInteTest, TestWithRowIdsSimple) {
CreateTable();
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields_);
std::vector<std::string> write_cols = {"f0", "f1", "f2"};
// row0: 0, "a0", "b0"; row1: 1, "a1", "b1" ...
auto src_array = PrepareBulkData(
1000,
[](int32_t i) {
return std::to_string(i) + ", \"a" + std::to_string(i) + "\", \"b" + std::to_string(i) +
"\"";
},
fields_);
ASSERT_OK_AND_ASSIGN(auto commit_msgs,
WriteArray(table_path, /*partition=*/{}, write_cols, {src_array}));
ASSERT_OK(Commit(table_path, commit_msgs));
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
[10, "a10", "b10"],
[999, "a999", "b999"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array,
/*predicate=*/nullptr,
/*row_ranges=*/{Range(999l, 999l), Range(10l, 10l)}));
}
TEST_P(BlobTableInteTest, TestWithRowIdsForMultipleBlobFiles) {
auto file_format = GetParam();
std::map<std::string, std::string> options = {
{Options::FILE_FORMAT, file_format}, {Options::TARGET_FILE_SIZE, "1000"},
{Options::BLOB_TARGET_FILE_SIZE, "80"}, {Options::BUCKET, "-1"},
{Options::ROW_TRACKING_ENABLED, "true"}, {Options::DATA_EVOLUTION_ENABLED, "true"},
{Options::FILE_SYSTEM, "local"}};
CreateTable(/*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields_);
// target blob size is 80, therefore, each 4 rows in blob will be in one file
std::vector<std::string> write_cols = {"f0", "f1", "f2"};
auto src_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
[0, "aaa0", "b0"],
[1, "aaa1", "b1"],
[2, "aaa2", "b2"],
[3, "aaa3", "b3"],
[4, "aaa4", "b4"],
[5, "aaa5", "b5"],
[6, "aaa6", "b6"],
[7, "aaa7", "b7"],
[8, "aaa8", "b8"],
[9, "aaa9", "b9"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs,
WriteArray(table_path, /*partition=*/{}, write_cols, {src_array}));
ASSERT_OK(Commit(table_path, commit_msgs));
{
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
[1, "aaa1", "b1"],
[8, "aaa8", "b8"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array,
/*predicate=*/nullptr,
/*row_ranges=*/{Range(1l, 1l), Range(8l, 8l)}));
}
{
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
[1, "aaa1", "b1"],
[5, "aaa5", "b5"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array,
/*predicate=*/nullptr,
/*row_ranges=*/{Range(1l, 1l), Range(5l, 5l)}));
}
{
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
[5, "aaa5", "b5"],
[6, "aaa6", "b6"],
[8, "aaa8", "b8"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array,
/*predicate=*/nullptr,
/*row_ranges=*/{Range(5l, 6l), Range(8l, 8l)}));
}
{
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
[5, "aaa5", "b5"],
[7, "aaa7", "b7"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array,
/*predicate=*/nullptr,
/*row_ranges=*/{Range(5l, 5l), Range(7l, 7l)}));
}
{
// f0 0-9 in one file, predicate does not skip any data file
auto predicate = PredicateBuilder::Equal(/*field_index=*/0, /*field_name=*/"f0",
FieldType::INT, Literal(5));
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
[1, "aaa1", "b1"],
[5, "aaa5", "b5"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array, predicate,
/*row_ranges=*/{Range(1l, 1l), Range(5l, 5l)}));
}
{
// test not read blob field
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[1]}), R"([
["aaa1"],
["aaa8"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, {"f1"}, expected_array,
/*predicate=*/nullptr,
/*row_ranges=*/{Range(1l, 1l), Range(8l, 8l)}));
}
}
TEST_P(BlobTableInteTest, TestAppendTableWriteWithMultipleBlobFields) {
arrow::FieldVector fields = {
arrow::field("f0", arrow::utf8()), arrow::field("f1", arrow::int32()),
BlobUtils::ToArrowField("blob1", true), BlobUtils::ToArrowField("blob2", true)};
std::map<std::string, std::string> options = {{Options::FILE_FORMAT, GetParam()},
{Options::TARGET_FILE_SIZE, "700"},
{Options::BUCKET, "-1"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
{Options::FILE_SYSTEM, "local"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
std::string data_json = R"([
["str_0", null, "apple", "red"],
["str_1", 1, "banana", "yellow"],
["str_2", 2, "cat", "black"]
])";
auto write_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), data_json).ValueOrDie());
auto schema = arrow::schema(fields);
ASSERT_OK_AND_ASSIGN(auto commit_msgs,
WriteArray(table_path, {}, schema->field_names(), {write_array}));
ASSERT_OK(Commit(table_path, commit_msgs));
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), write_array));
}
TEST_P(BlobTableInteTest, TestAppendWriteWithNullBlob) {
arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
BlobUtils::ToArrowField("blob", true)};
std::map<std::string, std::string> options = {{Options::FILE_FORMAT, GetParam()},
{Options::BUCKET, "-1"},
{Options::FILE_SYSTEM, "local"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
// Write: row 0 non-null blob, row 1 null blob, row 2 non-null blob
std::string data_json = R"([
[1, "hello"],
[2, null],
[3, "world"]
])";
auto write_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), data_json).ValueOrDie());
auto schema = arrow::schema(fields);
ASSERT_OK_AND_ASSIGN(auto commit_msgs,
WriteArray(table_path, {}, schema->field_names(), {write_array}));
ASSERT_OK(Commit(table_path, commit_msgs));
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), write_array));
}
TEST_P(BlobTableInteTest, TestReadTableWithMultiBlobFields) {
auto file_format = GetParam();
if (file_format == "avro" || file_format == "mosaic") {
return;
}
std::string table_path = paimon::test::GetDataDir() + file_format +
"/append_table_with_multi_blob.db/append_table_with_multi_blob";
std::vector<DataField> read_fields = {
DataField(5, BlobUtils::ToArrowField("f5")),
DataField(1, arrow::field("f1", arrow::int32())),
DataField(2, arrow::field("f2", arrow::int32())),
DataField(3, arrow::field("f3", arrow::float64())),
DataField(4, arrow::field("f4", arrow::utf8())),
SpecialFields::RowId(),
DataField(6, BlobUtils::ToArrowField("f6")),
};
std::shared_ptr<arrow::DataType> arrow_data_type =
DataField::ConvertDataFieldsToArrowStructType(read_fields);
auto make_json_row = [&](int32_t i) -> std::string {
std::string f5_json;
if (i <= 1) {
f5_json = "null";
} else {
f5_json = "\"" + std::string(1024, static_cast<char>('A' + i)) + "\"";
}
std::string f6_json;
if (i == 0 || i == 2) {
f6_json = "null";
} else {
f6_json = "\"" + std::string(2048, static_cast<char>('a' + i)) + "\"";
}
// f1=i, f2=i*10, f3=i+0.5, f4="desc_i", row_id=i
return "[" + f5_json + ", " + std::to_string(i) + ", " + std::to_string(i * 10) + ", " +
std::to_string(i + 0.5) + ", \"desc_" + std::to_string(i) + "\", " +
std::to_string(i) + ", " + f6_json + "]";
};
auto build_expected =
[&](const std::vector<int32_t>& row_indices) -> std::shared_ptr<arrow::StructArray> {
std::string json_str = "[";
for (size_t idx = 0; idx < row_indices.size(); idx++) {
if (idx > 0) {
json_str += ",";
}
json_str += make_json_row(row_indices[idx]);
}
json_str += "]";
return std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type, json_str).ValueOrDie());
};
// Full scan: all 10 rows
{
std::vector<int32_t> all_rows(10);
std::iota(all_rows.begin(), all_rows.end(), 0);
auto expected_array = build_expected(all_rows);
ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
expected_array));
}
// Test with row_ranges: select row 0 (both null), row 2 (f5 non-null, f6 null),
// row 5 (both non-null) to verify null handling under row range filtering.
{
auto expected_rr = build_expected({0, 2, 5});
ASSERT_OK(ScanAndRead(table_path, arrow::schema(arrow_data_type->fields())->field_names(),
expected_rr, /*predicate=*/nullptr,
/*row_ranges=*/{Range(0l, 0l), Range(2l, 2l), Range(5l, 5l)}));
}
}
TEST_P(BlobTableInteTest, TestBlobDescriptorField) {
if (GetParam() == "mosaic") {
return;
}
// Two blob fields configured via BLOB_DESCRIPTOR_FIELD and stored inline as descriptors.
arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
BlobUtils::ToArrowField("b0", true),
BlobUtils::ToArrowField("b1", true)};
std::map<std::string, std::string> options = {{Options::FILE_FORMAT, GetParam()},
{Options::TARGET_FILE_SIZE, "700"},
{Options::BUCKET, "-1"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
{Options::BLOB_DESCRIPTOR_FIELD, "b0,b1"},
{Options::FILE_SYSTEM, "local"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
// Input uses plain raw bytes for readability
std::string raw_json = R"([
[1, "image_data_0", "video_data_0"],
[2, "image_data_1", "video_data_1"],
[3, "image_data_2", "video_data_2"]
])";
auto raw_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), raw_json).ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto desc_array, ConvertRawBlobToDescriptor(raw_array, {"b0", "b1"}));
// write descriptor array
auto schema = arrow::schema(fields);
ASSERT_OK_AND_ASSIGN(auto commit_msgs,
WriteArray(table_path, {}, schema->field_names(), {desc_array}));
ASSERT_OK(Commit(table_path, commit_msgs));
// Scan and verify DataFileMeta: all blob fields are inline descriptors, so write_cols is unset.
ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
VerifyDataFileMetas(plan, /*expected_file_count=*/1, /*expected_row_counts=*/{3},
/*expected_min_seqs=*/{1}, /*expected_max_seqs=*/{1},
/*expected_first_row_ids=*/{0},
/*expected_write_cols=*/{std::nullopt});
// Read and resolve descriptors back to raw bytes
std::map<std::string, std::string> read_options = {};
ASSERT_OK_AND_ASSIGN(auto result, ReadTable(table_path, schema->field_names(), plan,
/*predicate=*/nullptr, read_options));
ASSERT_TRUE(result);
auto read_concat = arrow::Concatenate(result->chunks()).ValueOrDie();
auto read_struct = std::dynamic_pointer_cast<arrow::StructArray>(read_concat);
ASSERT_OK_AND_ASSIGN(auto resolved, ConvertDescriptorToRawBlob(read_struct, {"b0", "b1"}));
ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(raw_array));
ASSERT_TRUE(resolved->Equals(expected_with_rk));
// Descriptor bytes should be unchanged (inline, not repacked)
ASSERT_TRUE(read_struct->GetFieldByName("b0")->Equals(desc_array->GetFieldByName("b0")));
ASSERT_TRUE(read_struct->GetFieldByName("b1")->Equals(desc_array->GetFieldByName("b1")));
}
TEST_P(BlobTableInteTest, TestBlobDescriptorFieldPartialInline) {
if (GetParam() == "mosaic") {
return;
}
// 4 blob fields: b0,b1 are inline descriptors; b2,b3 are regular blob fields written to
// .blob files.
arrow::FieldVector fields = {
arrow::field("f0", arrow::int32()), BlobUtils::ToArrowField("b0", true),
BlobUtils::ToArrowField("b1", true), BlobUtils::ToArrowField("b2", true),
BlobUtils::ToArrowField("b3", true)};
std::map<std::string, std::string> options = {{Options::FILE_FORMAT, GetParam()},
{Options::TARGET_FILE_SIZE, "700"},
{Options::BUCKET, "-1"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
{Options::BLOB_DESCRIPTOR_FIELD, "b0,b1"},
{Options::FILE_SYSTEM, "local"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
// Input uses plain raw bytes:
// b0: all non-null, b1: has nulls, b2: all non-null, b3: has nulls
std::string raw_json = R"([
[1, "img_0", null, "raw_2_0", "raw_3_0"],
[2, "img_1", "vid_1", "raw_2_1", null ],
[3, "img_2", null, "raw_2_2", "raw_3_2" ]
])";
auto raw_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), raw_json).ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto desc_array,
ConvertRawBlobToDescriptor(raw_array, {"b0", "b1", "b2", "b3"}));
// write: b0,b1 as descriptor bytes; b2,b3 as raw bytes (paimon writes them to .blob files)
auto schema = arrow::schema(fields);
ASSERT_OK_AND_ASSIGN(auto commit_msgs,
WriteArray(table_path, {}, schema->field_names(), {desc_array}));
ASSERT_OK(Commit(table_path, commit_msgs));
// Scan and verify DataFileMeta: b2,b3 go to .blob files, "f0", "b0", "b1" go to main files.
ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
VerifyDataFileMetas(plan, /*expected_file_count=*/3, /*expected_row_counts=*/{3, 3, 3},
/*expected_min_seqs=*/{1, 1, 1}, /*expected_max_seqs=*/{1, 1, 1},
/*expected_first_row_ids=*/{0, 0, 0},
/*expected_write_cols=*/
{std::vector<std::string>{"f0", "b0", "b1"}, std::vector<std::string>{"b2"},
std::vector<std::string>{"b3"}});
std::map<std::string, std::string> read_options = {{Options::BLOB_AS_DESCRIPTOR, "true"}};
ASSERT_OK_AND_ASSIGN(auto result, ReadTable(table_path, schema->field_names(), plan,
/*predicate=*/nullptr, read_options));
ASSERT_TRUE(result);
auto read_concat = arrow::Concatenate(result->chunks()).ValueOrDie();
auto read_struct = std::dynamic_pointer_cast<arrow::StructArray>(read_concat);
// b0,b1 inline descriptor (not repacked), should match input
ASSERT_TRUE(read_struct->GetFieldByName("b0")->Equals(desc_array->GetFieldByName("b0")));
ASSERT_TRUE(read_struct->GetFieldByName("b1")->Equals(desc_array->GetFieldByName("b1")));
// Resolve b0,b1 descriptors back to raw bytes, then compare full struct
ASSERT_OK_AND_ASSIGN(auto resolved,
ConvertDescriptorToRawBlob(read_struct, {"b0", "b1", "b2", "b3"}));
ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(raw_array));
ASSERT_TRUE(resolved->Equals(expected_with_rk));
}
TEST_P(BlobTableInteTest, TestBlobDescriptorMultiCommitAndShuffledReadSchema) {
if (GetParam() == "mosaic") {
return;
}
// Multiple write+commit rounds with a shuffled read schema: b3, b2, b1, b0, f0.
arrow::FieldVector fields = {
arrow::field("f0", arrow::int32()), BlobUtils::ToArrowField("b0", true),
BlobUtils::ToArrowField("b1", true), BlobUtils::ToArrowField("b2", true),
BlobUtils::ToArrowField("b3", true)};
std::map<std::string, std::string> options = {{Options::FILE_FORMAT, GetParam()},
{Options::TARGET_FILE_SIZE, "700"},
{Options::BUCKET, "-1"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
{Options::BLOB_DESCRIPTOR_FIELD, "b0,b1"},
{Options::FILE_SYSTEM, "local"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields);
// --- First write+commit ---
std::string raw_json_1 = R"([
[1, "img_0", null, "raw_2_0", "raw_3_0"],
[2, "img_1", "vid_1", "raw_2_1", null ]
])";
auto raw_array_1 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), raw_json_1).ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto desc_array_1, ConvertRawBlobToDescriptor(raw_array_1, {"b0", "b1"}));
ASSERT_OK_AND_ASSIGN(auto commit_msgs_1,
WriteArray(table_path, {}, schema->field_names(), {desc_array_1}));
ASSERT_OK(Commit(table_path, commit_msgs_1));
// --- Second write+commit ---
std::string raw_json_2 = R"([
[3, "img_2", "vid_2", "raw_2_2", "raw_3_2"],
[4, null, "vid_3", "raw_2_3", "raw_3_3"]
])";
auto raw_array_2 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), raw_json_2).ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto desc_array_2, ConvertRawBlobToDescriptor(raw_array_2, {"b0", "b1"}));
ASSERT_OK_AND_ASSIGN(auto commit_msgs_2,
WriteArray(table_path, {}, schema->field_names(), {desc_array_2}));
ASSERT_OK(Commit(table_path, commit_msgs_2));
// --- Third write+commit ---
std::string raw_json_3 = R"([
[5, "img_4", null, "raw_2_4", null ],
[6, "img_5", "vid_5", null, "raw_3_5"]
])";
auto raw_array_3 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), raw_json_3).ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto desc_array_3, ConvertRawBlobToDescriptor(raw_array_3, {"b0", "b1"}));
ASSERT_OK_AND_ASSIGN(auto commit_msgs_3,
WriteArray(table_path, {}, schema->field_names(), {desc_array_3}));
ASSERT_OK(Commit(table_path, commit_msgs_3));
// test read
{
// --- Read with shuffled schema: b3, b2, b1, b0, f0 ---
std::vector<std::string> shuffled_read_schema = {"b3", "b2", "b1", "b0", "f0"};
ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
std::map<std::string, std::string> read_options = {{Options::BLOB_AS_DESCRIPTOR, "false"}};
ASSERT_OK_AND_ASSIGN(auto result, ReadTable(table_path, shuffled_read_schema, plan,
/*predicate=*/nullptr, read_options));
ASSERT_TRUE(result);
auto read_concat = arrow::Concatenate(result->chunks()).ValueOrDie();
auto read_struct = std::dynamic_pointer_cast<arrow::StructArray>(read_concat);
// Build expected array in shuffled order from all 3 batches
arrow::FieldVector shuffled_fields = {
BlobUtils::ToArrowField("b3", true), BlobUtils::ToArrowField("b2", true),
BlobUtils::ToArrowField("b1", true), BlobUtils::ToArrowField("b0", true),
arrow::field("f0", arrow::int32())};
std::string expected_json = R"([
["raw_3_0", "raw_2_0", null, "img_0", 1],
[null, "raw_2_1", "vid_1", "img_1", 2],
["raw_3_2", "raw_2_2", "vid_2", "img_2", 3],
["raw_3_3", "raw_2_3", "vid_3", null, 4],
[null, "raw_2_4", null, "img_4", 5],
["raw_3_5", null, "vid_5", "img_5", 6]
])";
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(shuffled_fields),
expected_json)
.ValueOrDie());
// Resolve descriptors (b0, b1 are descriptor fields) back to raw bytes
ASSERT_OK_AND_ASSIGN(auto resolved, ConvertDescriptorToRawBlob(read_struct, {"b0", "b1"}));
ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(expected_array));
ASSERT_TRUE(resolved->Equals(expected_with_rk));
}
{
// test scan and read with GlobalIndexResult
std::vector<std::string> shuffled_read_schema = {"b3", "b2", "b1", "b0", "f0"};
ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path, /*predicate=*/nullptr,
/*row_ranges=*/{Range(1, 3), Range(5, 5)}));
std::map<std::string, std::string> read_options = {{Options::BLOB_AS_DESCRIPTOR, "false"}};
ASSERT_OK_AND_ASSIGN(auto result, ReadTable(table_path, shuffled_read_schema, plan,
/*predicate=*/nullptr, read_options));
ASSERT_TRUE(result);
auto read_concat = arrow::Concatenate(result->chunks()).ValueOrDie();
auto read_struct = std::dynamic_pointer_cast<arrow::StructArray>(read_concat);
// Build expected array in shuffled order from all 3 batches
arrow::FieldVector shuffled_fields = {
BlobUtils::ToArrowField("b3", true), BlobUtils::ToArrowField("b2", true),
BlobUtils::ToArrowField("b1", true), BlobUtils::ToArrowField("b0", true),
arrow::field("f0", arrow::int32())};
std::string expected_json = R"([
[null, "raw_2_1", "vid_1", "img_1", 2],
["raw_3_2", "raw_2_2", "vid_2", "img_2", 3],
["raw_3_3", "raw_2_3", "vid_3", null, 4],
["raw_3_5", null, "vid_5", "img_5", 6]
])";
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(shuffled_fields),
expected_json)
.ValueOrDie());
// Resolve descriptors (b0, b1 are descriptor fields) back to raw bytes
ASSERT_OK_AND_ASSIGN(auto resolved, ConvertDescriptorToRawBlob(read_struct, {"b0", "b1"}));
ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(expected_array));
ASSERT_TRUE(resolved->Equals(expected_with_rk));
}
}
// The shared-shredding map is read from one main data file while the blob payload is read from a
// separate blob file with the same row-id range.
TEST_P(BlobTableInteTest, TestSharedShreddingWithBlobDataEvolution) {
if (GetParam() == "avro" || GetParam() == "mosaic") {
return;
}
arrow::FieldVector fields = {
arrow::field("id", arrow::int32()),
arrow::field("tags", arrow::map(arrow::utf8(), arrow::int64())),
BlobUtils::ToArrowField("payload"),
};
std::map<std::string, std::string> options = {
{Options::FILE_FORMAT, GetParam()},
{Options::BUCKET, "-1"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
{Options::FILE_SYSTEM, "local"},
{"fields.tags.map.storage-layout", "shared-shredding"},
{"fields.tags.map.shared-shredding.max-columns", "1"},
};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto map_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[0], fields[1]}),
R"([
[1, [["a", 10], ["z", 11]]],
[2, [["a", 20]]],
[3, null]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto map_msgs, WriteArray(table_path, {}, {"id", "tags"}, {map_array}));
ASSERT_OK(Commit(table_path, map_msgs));
auto blob_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[0], fields[2]}),
R"([
[1, "payload-1"],
[2, "payload-2"],
[3, "payload-3"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto blob_msgs,
WriteArray(table_path, {}, {"id", "payload"}, {blob_array}));
SetFirstRowId(0, blob_msgs);
ASSERT_OK(Commit(table_path, blob_msgs));
auto expected = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields),
R"([
[1, [["a", 10], ["z", 11]], "payload-1"],
[2, [["a", 20]], "payload-2"],
[3, null, "payload-3"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, arrow::schema(fields)->field_names(), expected));
}
// Two independent shared-shredding map columns are written into different main files.
TEST_P(BlobTableInteTest, TestMultipleSharedShreddingMapsWithBlobDataEvolution) {
if (GetParam() == "avro" || GetParam() == "mosaic") {
return;
}
arrow::FieldVector fields = {
arrow::field("id", arrow::int32()),
arrow::field("f0", arrow::map(arrow::utf8(), arrow::int64())),
arrow::field("f1", arrow::map(arrow::utf8(), arrow::utf8())),
BlobUtils::ToArrowField("payload"),
};
std::map<std::string, std::string> options = {
{Options::FILE_FORMAT, GetParam()},
{Options::BUCKET, "-1"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
{Options::FILE_SYSTEM, "local"},
{"fields.f0.map.storage-layout", "shared-shredding"},
{"fields.f0.map.shared-shredding.max-columns", "1"},
{"fields.f1.map.storage-layout", "shared-shredding"},
{"fields.f1.map.shared-shredding.max-columns", "1"},
};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto f0_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[0], fields[1]}),
R"([
[1, [["a", 10], ["z", 11]]],
[2, [["a", 20]]],
[3, null]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto f0_msgs, WriteArray(table_path, {}, {"id", "f0"}, {f0_array}));
ASSERT_OK(Commit(table_path, f0_msgs));
auto f1_blob_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[2], fields[3]}),
R"([
[[["b", "red"], ["y", "blue"]], "payload-1"],
[[["b", "green"]], "payload-2"],
[[], "payload-3"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto f1_blob_msgs,
WriteArray(table_path, {}, {"f1", "payload"}, {f1_blob_array}));
SetFirstRowId(0, f1_blob_msgs);
ASSERT_OK(Commit(table_path, f1_blob_msgs));
auto expected = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields),
R"([
[1, [["a", 10], ["z", 11]], [["b", "red"], ["y", "blue"]], "payload-1"],
[2, [["a", 20]], [["b", "green"]], "payload-2"],
[3, null, [], "payload-3"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, arrow::schema(fields)->field_names(), expected));
}
// A newer partial data file rewrites only the shared-shredding map for the same row-id range.
TEST_P(BlobTableInteTest, TestSharedShreddingMapOverrideWithBlobDataEvolution) {
if (GetParam() == "avro" || GetParam() == "mosaic") {
return;
}
arrow::FieldVector fields = {
arrow::field("id", arrow::int32()),
arrow::field("tags", arrow::map(arrow::utf8(), arrow::int64())),
BlobUtils::ToArrowField("payload"),
};
std::map<std::string, std::string> options = {
{Options::FILE_FORMAT, GetParam()},
{Options::BUCKET, "-1"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
{Options::FILE_SYSTEM, "local"},
{"fields.tags.map.storage-layout", "shared-shredding"},
{"fields.tags.map.shared-shredding.max-columns", "1"},
};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto old_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields),
R"([
[1, [["a", 10], ["z", 11]], "payload-1"],
[2, [["a", 20]], "payload-2"],
[3, [["a", 30]], "payload-3"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(
auto old_msgs,
WriteArray(table_path, {}, arrow::schema(fields)->field_names(), {old_array}));
ASSERT_OK(Commit(table_path, old_msgs));
auto new_map_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[1]}),
R"([
[[["a", 100], ["z", 101]]],
[null],
[[]]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto new_map_msgs, WriteArray(table_path, {}, {"tags"}, {new_map_array}));
SetFirstRowId(0, new_map_msgs);
ASSERT_OK(Commit(table_path, new_map_msgs));
auto expected = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields),
R"([
[1, [["a", 100], ["z", 101]], "payload-1"],
[2, null, "payload-2"],
[3, [], "payload-3"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, arrow::schema(fields)->field_names(), expected));
}
TEST_P(BlobTableInteTest, TestOrcMapStorageLayoutEvolutionWithBlobDataEvolution) {
if (GetParam() != "orc") {
return;
}
arrow::FieldVector fields = {
arrow::field("id", arrow::int32()),
arrow::field("tags", arrow::map(arrow::utf8(), arrow::utf8())),
BlobUtils::ToArrowField("payload"),
};
std::map<std::string, std::string> options_v0 = {
{Options::FILE_FORMAT, "orc"},
{Options::BUCKET, "-1"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
{Options::FILE_SYSTEM, "local"},
{"fields.tags.map.storage-layout", "default"},
{"orc.read.enable-lazy-decoding", "true"},
{"orc.dictionary-key-size-threshold", "1"},
};
CreateTable(fields, /*partition_keys=*/{}, options_v0);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto array_v0 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields),
R"([
[1, [["a", "red"], ["z", "blue"]], "payload-1"],
[2, [["a", "red"], ["z", "green"]], "payload-2"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(
auto msgs_v0, WriteArray(table_path, {}, arrow::schema(fields)->field_names(), {array_v0}));
ASSERT_OK(Commit(table_path, msgs_v0));
std::map<std::string, std::string> options_v1 = options_v0;
options_v1["fields.tags.map.storage-layout"] = "shared-shredding";
options_v1["fields.tags.map.shared-shredding.max-columns"] = "1";
ASSERT_OK(WriteNextSchema(
table_path, {DataField(0, fields[0]), DataField(1, fields[1]), DataField(2, fields[2])},
/*highest_field_id=*/2, options_v1));
auto array_v1 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields),
R"([
[3, [["a", "red"], ["z", "yellow"]], "payload-3"],
[4, [["a", "red"]], "payload-4"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(
auto msgs_v1, WriteArray(table_path, {}, arrow::schema(fields)->field_names(), {array_v1}));
ASSERT_OK(Commit(table_path, msgs_v1));
auto expected = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields),
R"([
[1, [["a", "red"], ["z", "blue"]], "payload-1"],
[2, [["a", "red"], ["z", "green"]], "payload-2"],
[3, [["a", "red"], ["z", "yellow"]], "payload-3"],
[4, [["a", "red"]], "payload-4"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, arrow::schema(fields)->field_names(), expected));
}
TEST_P(BlobTableInteTest, TestDataEvolutionWithBlobDescriptorField) {
if (GetParam() == "mosaic") {
return;
}
// Test DataEvolution (split-column write) combined with blob descriptor fields.
// Schema: f0(int32), b0/b1(blob descriptor inline), b2/b3(blob).
// Commit 1: file A writes (f0, b2, b3)
// Commit 2: file B writes (f0, b0, b1) with SetFirstRowId(0)
// -> merges with commit 1
// Commit 3: file A writes (f0, b0, b1, b3)
// Commit 4: file B writes (b0, b1, b3) with SetFirstRowId(3)
// -> merges with commit 3
arrow::FieldVector fields = {
arrow::field("f0", arrow::int32()), BlobUtils::ToArrowField("b0", true),
BlobUtils::ToArrowField("b1", true), BlobUtils::ToArrowField("b2", true),
BlobUtils::ToArrowField("b3", true)};
std::map<std::string, std::string> options = {{Options::FILE_FORMAT, GetParam()},
{Options::TARGET_FILE_SIZE, "700"},
{Options::BUCKET, "-1"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
{Options::BLOB_DESCRIPTOR_FIELD, "b0,b1"},
{Options::FILE_SYSTEM, "local"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
// --- Commit 1: file A (f0, b2, b3), Commit 2: file B (f0, b0, b1) SetFirstRowId(0) ---
std::string file_a1_json = R"([
[1, "raw_2_0", "raw_3_0"],
[2, "raw_2_1", null ],
[3, null, "raw_3_2"]
])";
arrow::FieldVector file_a1_fields = {fields[0], fields[3], fields[4]};
auto file_a1_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(file_a1_fields), file_a1_json)
.ValueOrDie());
std::string file_b1_json = R"([
[1, "img_0", "vid_0"],
[2, "img_1", null ],
[3, "img_2", "vid_2"]
])";
arrow::FieldVector file_b1_fields = {fields[0], fields[1], fields[2]};
auto file_b1_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(file_b1_fields), file_b1_json)
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto file_b1_desc,
ConvertRawBlobToDescriptor(file_b1_array, {"b0", "b1"}));
ASSERT_OK_AND_ASSIGN(auto commit_msgs_a1,
WriteArray(table_path, {}, {"f0", "b2", "b3"}, {file_a1_array}));
ASSERT_OK(Commit(table_path, commit_msgs_a1));
ASSERT_OK_AND_ASSIGN(auto commit_msgs_b1,
WriteArray(table_path, {}, {"f0", "b0", "b1"}, {file_b1_desc}));
SetFirstRowId(0, commit_msgs_b1);
ASSERT_OK(Commit(table_path, commit_msgs_b1));
// --- Commit 3: file A (f0, b0, b1, b3), Commit 4: file B (b0, b1, b3) SetFirstRowId(3) ---
// Duplicate cols b0, b1, b3: file B (commit 4, newer) takes precedence.
std::string file_a2_json = R"([
[4, "img_3_old", "vid_3_old", "raw_3_3_old"],
[5, null, "vid_4_old", "raw_3_4_old"],
[6, "img_5_old", null, null ]
])";
arrow::FieldVector file_a2_fields = {fields[0], fields[1], fields[2], fields[4]};
auto file_a2_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(file_a2_fields), file_a2_json)
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto file_a2_desc,
ConvertRawBlobToDescriptor(file_a2_array, {"b0", "b1"}));
std::string file_b2_json = R"([
["img_3", "vid_3", "raw_3_3"],
[null, "vid_4", "raw_3_4"],
["img_5", null, null ]
])";
arrow::FieldVector file_b2_fields = {fields[1], fields[2], fields[4]};
auto file_b2_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(file_b2_fields), file_b2_json)
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto file_b2_desc,
ConvertRawBlobToDescriptor(file_b2_array, {"b0", "b1"}));
ASSERT_OK_AND_ASSIGN(auto commit_msgs_a2,
WriteArray(table_path, {}, {"f0", "b0", "b1", "b3"}, {file_a2_desc}));
ASSERT_OK(Commit(table_path, commit_msgs_a2));
ASSERT_OK_AND_ASSIGN(auto commit_msgs_b2,
WriteArray(table_path, {}, {"b0", "b1", "b3"}, {file_b2_desc}));
SetFirstRowId(3, commit_msgs_b2);
ASSERT_OK(Commit(table_path, commit_msgs_b2));
// --- Read all data with full schema ---
std::vector<std::string> read_schema = {"f0", "b0", "b1", "b2", "b3"};
ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
std::map<std::string, std::string> read_options = {{Options::BLOB_AS_DESCRIPTOR, "false"}};
ASSERT_OK_AND_ASSIGN(auto result, ReadTable(table_path, read_schema, plan,
/*predicate=*/nullptr, read_options));
ASSERT_TRUE(result);
auto read_concat = arrow::Concatenate(result->chunks()).ValueOrDie();
auto read_struct = std::dynamic_pointer_cast<arrow::StructArray>(read_concat);
ASSERT_EQ(read_struct->length(), 6);
// Expected: round1 all columns present; round2 b2=null, b0/b1/b3 from file B (newer)
std::string expected_json = R"([
[1, "img_0", "vid_0", "raw_2_0", "raw_3_0"],
[2, "img_1", null, "raw_2_1", null ],
[3, "img_2", "vid_2", null, "raw_3_2" ],
[4, "img_3", "vid_3", null, "raw_3_3" ],
[5, null, "vid_4", null, "raw_3_4" ],
[6, "img_5", null, null, null ]
])";
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), expected_json)
.ValueOrDie());
// Resolve descriptors back to raw bytes
ASSERT_OK_AND_ASSIGN(auto resolved, ConvertDescriptorToRawBlob(read_struct, {"b0", "b1"}));
ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(expected_array));
ASSERT_TRUE(resolved->type()->Equals(expected_with_rk->type()));
ASSERT_TRUE(resolved->Equals(expected_with_rk));
}
TEST_P(BlobTableInteTest, TestBlobDescriptorFieldWriteRawBytesDirectly) {
if (GetParam() == "mosaic") {
return;
}
// Similar to TestBlobDescriptorField but writes raw bytes directly without converting to
// descriptor first. Descriptor fields reject values without the descriptor magic header.
arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
BlobUtils::ToArrowField("b0", true),
BlobUtils::ToArrowField("b1", true)};
std::map<std::string, std::string> options = {{Options::FILE_FORMAT, GetParam()},
{Options::TARGET_FILE_SIZE, "700"},
{Options::BUCKET, "-1"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
{Options::BLOB_DESCRIPTOR_FIELD, "b0,b1"},
{Options::FILE_SYSTEM, "local"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
// Write raw bytes directly (no ConvertRawBlobToDescriptor)
std::string raw_json = R"([
[1, "image_data_0", "video_data_0"],
[2, "image_data_1", "video_data_1"],
[3, "image_data_2", "video_data_2"]
])";
auto raw_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), raw_json).ValueOrDie());
auto schema = arrow::schema(fields);
ASSERT_NOK_WITH_MSG(WriteArray(table_path, {}, schema->field_names(), {raw_array}),
"BLOB inline field b0 require values to be set as corresponding type.");
}
TEST_P(BlobTableInteTest, TestBlobViewFieldWithUpstreamTable) {
auto file_format = GetParam();
if (file_format == "avro" || file_format == "mosaic") {
return;
}
const std::string upstream_db_name = "append_table_with_multi_blob";
const std::string upstream_table_name = "append_table_with_multi_blob";
std::string src_db_path = paimon::test::GetDataDir() + file_format + "/" + upstream_db_name +
".db/" + upstream_table_name;
std::string dst_db_path =
PathUtil::JoinPath(dir_->Str(), upstream_db_name + ".db/" + upstream_table_name);
ASSERT_TRUE(TestUtil::CopyDirectory(src_db_path, dst_db_path));
arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
BlobUtils::ToArrowField("view", true)};
std::map<std::string, std::string> options = {
{Options::FILE_FORMAT, file_format}, {Options::BUCKET, "-1"},
{Options::ROW_TRACKING_ENABLED, "true"}, {Options::DATA_EVOLUTION_ENABLED, "true"},
{Options::BLOB_VIEW_FIELD, "view"}, {Options::FILE_SYSTEM, "local"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
// src array
Identifier upstream_identifier(upstream_db_name, upstream_table_name);
arrow::LargeBinaryBuilder view_builder;
for (int32_t i = 0; i < 8; ++i) {
if (i < 6) {
BlobViewStruct view_struct(upstream_identifier, /*field_id=*/6,
/*row_id=*/static_cast<int64_t>(i));
ASSERT_OK_AND_ASSIGN(PAIMON_UNIQUE_PTR<Bytes> serialized, view_struct.Serialize(pool_));
ASSERT_TRUE(view_builder
.Append(reinterpret_cast<const uint8_t*>(serialized->data()),
serialized->size())
.ok());
} else {
ASSERT_TRUE(view_builder.AppendNull().ok());
}
}
std::shared_ptr<arrow::Array> write_view_array;
ASSERT_TRUE(view_builder.Finish(&write_view_array).ok());
auto write_f0_array = arrow::ipc::internal::json::ArrayFromJSON(
arrow::int32(), R"([100,101,102,103,104,105,106,107])")
.ValueOrDie();
auto write_struct = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::StructArray::Make(arrow::ArrayVector({write_f0_array, write_view_array}),
std::vector<std::string>({"f0", "view"}))
.ValueOrDie());
// write & commit
auto schema = arrow::schema(fields);
ASSERT_OK_AND_ASSIGN(auto commit_msgs,
WriteArray(table_path, {}, schema->field_names(), {write_struct}));
ASSERT_OK(Commit(table_path, commit_msgs));
std::string padding_b(2048, 'b');
std::string padding_d(2048, 'd');
std::string padding_e(2048, 'e');
std::string padding_f(2048, 'f');
// scan & read
ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
ASSERT_OK_AND_ASSIGN(auto result,
ReadTable(table_path, schema->field_names(), plan, /*predicate=*/nullptr));
ASSERT_TRUE(result);
auto read_concat = arrow::Concatenate(result->chunks()).ValueOrDie();
auto read_struct = std::dynamic_pointer_cast<arrow::StructArray>(read_concat);
ASSERT_EQ(read_struct->length(), 8);
ASSERT_OK_AND_ASSIGN(auto result_array, ConvertDescriptorToRawBlob(read_struct, {"view"}));
// clang-format off
std::string expected_json = R"([
[100, null],
[101, ")" + padding_b + R"("],
[102, null],
[103, ")" + padding_d + R"("],
[104, ")" + padding_e + R"("],
[105, ")" + padding_f + R"("],
[106, null],
[107, null]
])";
// clang-format on
auto expected_struct = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), expected_json)
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(expected_struct));
ASSERT_TRUE(result_array->Equals(expected_with_rk))
<< "result_array:" << result_array->ToString() << std::endl
<< "expected:" << expected_with_rk->ToString();
// Sub-case 1: scan with row_ranges.
{
ASSERT_OK_AND_ASSIGN(auto range_plan, ScanTable(table_path, /*predicate=*/nullptr,
/*row_ranges=*/{Range(1, 3), Range(5, 5)}));
ASSERT_OK_AND_ASSIGN(auto range_result, ReadTable(table_path, schema->field_names(),
range_plan, /*predicate=*/nullptr));
ASSERT_TRUE(range_result);
auto range_concat = arrow::Concatenate(range_result->chunks()).ValueOrDie();
auto range_struct = std::dynamic_pointer_cast<arrow::StructArray>(range_concat);
ASSERT_EQ(range_struct->length(), 4);
ASSERT_OK_AND_ASSIGN(auto range_resolved,
ConvertDescriptorToRawBlob(range_struct, {"view"}));
// clang-format off
std::string range_json = R"([
[101, ")" + padding_b + R"("],
[102, null],
[103, ")" + padding_d + R"("],
[105, ")" + padding_f + R"("]
])";
// clang-format on
auto range_expected_struct = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), range_json)
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto range_expected_with_rk,
PrependRowKindColumn(range_expected_struct));
ASSERT_TRUE(range_resolved->Equals(range_expected_with_rk))
<< "range_resolved:" << range_resolved->ToString() << std::endl
<< "expected:" << range_expected_with_rk->ToString();
}
// Sub-case 2: scan with predicate (f0 > 102), data evolution split read will ignore format push
// down
{
auto predicate = PredicateBuilder::GreaterThan(/*field_index=*/0, /*field_name=*/"f0",
FieldType::INT, Literal(102));
ASSERT_OK_AND_ASSIGN(auto pred_plan, ScanTable(table_path, predicate, /*row_ranges=*/{}));
ASSERT_OK_AND_ASSIGN(auto pred_result,
ReadTable(table_path, schema->field_names(), pred_plan, predicate));
ASSERT_TRUE(pred_result);
auto pred_concat = arrow::Concatenate(pred_result->chunks()).ValueOrDie();
auto pred_struct = std::dynamic_pointer_cast<arrow::StructArray>(pred_concat);
ASSERT_EQ(pred_struct->length(), 8);
ASSERT_OK_AND_ASSIGN(auto pred_resolved, ConvertDescriptorToRawBlob(pred_struct, {"view"}));
// clang-format off
std::string pred_json = R"([
[100, null],
[101, ")" + padding_b + R"("],
[102, null],
[103, ")" + padding_d + R"("],
[104, ")" + padding_e + R"("],
[105, ")" + padding_f + R"("],
[106, null],
[107, null]
])";
// clang-format on
auto pred_expected_struct = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), pred_json)
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto pred_expected_with_rk,
PrependRowKindColumn(pred_expected_struct));
ASSERT_TRUE(pred_resolved->Equals(pred_expected_with_rk))
<< "pred_resolved:" << pred_resolved->ToString() << std::endl
<< "expected:" << pred_expected_with_rk->ToString();
}
}
TEST_P(BlobTableInteTest, TestForwardBlobViewReference) {
auto file_format = GetParam();
if (file_format == "avro" || file_format == "mosaic") {
return;
}
// Forward blob view references between two blob-view tables: read the source table with
// resolve dynamically disabled, write the preserved BlobViewStruct bytes into the target
// table, then verify the target still stores the original upstream references and a
// default read resolves them to the actual upstream blob values.
const std::string upstream_db_name = "append_table_with_multi_blob";
const std::string upstream_table_name = "append_table_with_multi_blob";
std::string src_db_path = paimon::test::GetDataDir() + file_format + "/" + upstream_db_name +
".db/" + upstream_table_name;
std::string dst_db_path =
PathUtil::JoinPath(dir_->Str(), upstream_db_name + ".db/" + upstream_table_name);
ASSERT_TRUE(TestUtil::CopyDirectory(src_db_path, dst_db_path));
// Both blob-view tables resolve the upstream table from the warehouse they share.
arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
BlobUtils::ToArrowField("view", true)};
std::map<std::string, std::string> source_options = {
{Options::FILE_FORMAT, file_format}, {Options::BUCKET, "-1"},
{Options::ROW_TRACKING_ENABLED, "true"}, {Options::DATA_EVOLUTION_ENABLED, "true"},
{Options::BLOB_VIEW_FIELD, "view"}, {Options::FILE_SYSTEM, "local"}};
CreateTable(fields, /*partition_keys=*/{}, source_options);
std::string source_table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields);
::ArrowSchema c_target_schema;
ASSERT_TRUE(arrow::ExportSchema(*schema, &c_target_schema).ok());
ASSERT_OK_AND_ASSIGN(auto catalog, Catalog::Create(dir_->Str(), {}));
ASSERT_OK(catalog->CreateTable(Identifier("foo", "bar_forward"), &c_target_schema,
/*partition_keys=*/{}, /*primary_keys=*/{}, source_options,
/*ignore_if_exists=*/false));
std::string target_table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar_forward");
// src array
Identifier upstream_identifier(upstream_db_name, upstream_table_name);
arrow::LargeBinaryBuilder view_builder;
for (int32_t i = 0; i < 8; ++i) {
if (i < 6) {
BlobViewStruct view_struct(upstream_identifier, /*field_id=*/6,
/*row_id=*/static_cast<int64_t>(i));
ASSERT_OK_AND_ASSIGN(PAIMON_UNIQUE_PTR<Bytes> serialized, view_struct.Serialize(pool_));
ASSERT_TRUE(view_builder
.Append(reinterpret_cast<const uint8_t*>(serialized->data()),
serialized->size())
.ok());
} else {
ASSERT_TRUE(view_builder.AppendNull().ok());
}
}
std::shared_ptr<arrow::Array> write_view_array;
ASSERT_TRUE(view_builder.Finish(&write_view_array).ok());
auto write_f0_array = arrow::ipc::internal::json::ArrayFromJSON(
arrow::int32(), R"([100,101,102,103,104,105,106,107])")
.ValueOrDie();
auto write_struct = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::StructArray::Make(arrow::ArrayVector({write_f0_array, write_view_array}),
std::vector<std::string>({"f0", "view"}))
.ValueOrDie());
// write & commit into the source table
ASSERT_OK_AND_ASSIGN(auto commit_msgs,
WriteArray(source_table_path, {}, schema->field_names(), {write_struct}));
ASSERT_OK(Commit(source_table_path, commit_msgs));
// A default read resolves the copied upstream table in the source table's warehouse. The
// pass-through read below uses the dynamic option to preserve the raw BlobViewStruct bytes.
ASSERT_OK_AND_ASSIGN(auto source_plan, ScanTable(source_table_path));
ASSERT_OK_AND_ASSIGN(
auto source_default_result,
ReadTable(source_table_path, schema->field_names(), source_plan, /*predicate=*/nullptr));
ASSERT_TRUE(source_default_result);
ASSERT_EQ(source_default_result->length(), 8);
ASSERT_OK_AND_ASSIGN(
auto source_result,
ReadTable(source_table_path, schema->field_names(), source_plan, /*predicate=*/nullptr,
{{Options::BLOB_VIEW_RESOLVE_ENABLED, "false"}}));
ASSERT_TRUE(source_result);
auto source_concat = arrow::Concatenate(source_result->chunks()).ValueOrDie();
auto source_struct = std::dynamic_pointer_cast<arrow::StructArray>(source_concat);
ASSERT_EQ(source_struct->length(), 8);
auto forward_f0_array = source_struct->GetFieldByName("f0");
ASSERT_TRUE(forward_f0_array);
auto forward_view_array = source_struct->GetFieldByName("view");
ASSERT_TRUE(forward_view_array);
ASSERT_TRUE(forward_view_array->Equals(write_view_array))
<< "source view:" << forward_view_array->ToString() << std::endl
<< "written view:" << write_view_array->ToString();
// Forward the preserved references into the target blob-view table.
auto forward_struct = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::StructArray::Make(arrow::ArrayVector({forward_f0_array, forward_view_array}),
std::vector<std::string>({"f0", "view"}))
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(
auto forward_commit_msgs,
WriteArray(target_table_path, {}, schema->field_names(), {forward_struct}));
ASSERT_OK(Commit(target_table_path, forward_commit_msgs));
// The target table stores the original upstream references byte-identically.
ASSERT_OK_AND_ASSIGN(auto target_plan, ScanTable(target_table_path));
ASSERT_OK_AND_ASSIGN(
auto raw_target_result,
ReadTable(target_table_path, schema->field_names(), target_plan, /*predicate=*/nullptr,
{{Options::BLOB_VIEW_RESOLVE_ENABLED, "false"}}));
ASSERT_TRUE(raw_target_result);
auto raw_target_concat = arrow::Concatenate(raw_target_result->chunks()).ValueOrDie();
auto raw_target_struct = std::dynamic_pointer_cast<arrow::StructArray>(raw_target_concat);
ASSERT_EQ(raw_target_struct->length(), 8);
auto raw_target_view_array = raw_target_struct->GetFieldByName("view");
ASSERT_TRUE(raw_target_view_array);
ASSERT_TRUE(raw_target_view_array->Equals(write_view_array))
<< "target view:" << raw_target_view_array->ToString() << std::endl
<< "written view:" << write_view_array->ToString();
// A default read of the target table resolves to the actual upstream blob values.
ASSERT_OK_AND_ASSIGN(auto resolved_result,
ReadTable(target_table_path, schema->field_names(), target_plan,
/*predicate=*/nullptr));
ASSERT_TRUE(resolved_result);
auto resolved_concat = arrow::Concatenate(resolved_result->chunks()).ValueOrDie();
auto resolved_struct = std::dynamic_pointer_cast<arrow::StructArray>(resolved_concat);
ASSERT_EQ(resolved_struct->length(), 8);
ASSERT_OK_AND_ASSIGN(auto resolved, ConvertDescriptorToRawBlob(resolved_struct, {"view"}));
std::string padding_b(2048, 'b');
std::string padding_d(2048, 'd');
std::string padding_e(2048, 'e');
std::string padding_f(2048, 'f');
// clang-format off
std::string expected_json = R"([
[100, null],
[101, ")" + padding_b + R"("],
[102, null],
[103, ")" + padding_d + R"("],
[104, ")" + padding_e + R"("],
[105, ")" + padding_f + R"("],
[106, null],
[107, null]
])";
// clang-format on
auto expected_struct = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), expected_json)
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(expected_struct));
ASSERT_TRUE(resolved->Equals(expected_with_rk))
<< "resolved:" << resolved->ToString() << std::endl
<< "expected:" << expected_with_rk->ToString();
}
TEST_P(BlobTableInteTest, TestBlobViewFieldWithUpstreamDescriptorBlob) {
auto file_format = GetParam();
if (file_format == "mosaic") {
return;
}
// Upstream table has two blob descriptor fields. The downstream view references cells from
// both b0 (field_id=1) and b1 (field_id=2).
const std::string upstream_db_name = "upstream_two_blob";
const std::string upstream_table_name = "upstream_two_blob";
arrow::FieldVector upstream_fields = {arrow::field("f0", arrow::int32()),
BlobUtils::ToArrowField("b0", true),
BlobUtils::ToArrowField("b1", true)};
auto upstream_schema = arrow::schema(upstream_fields);
std::map<std::string, std::string> upstream_options = {
{Options::FILE_FORMAT, file_format}, {Options::BUCKET, "-1"},
{Options::ROW_TRACKING_ENABLED, "true"}, {Options::DATA_EVOLUTION_ENABLED, "true"},
{Options::BLOB_DESCRIPTOR_FIELD, "b0,b1"}, {Options::FILE_SYSTEM, "local"}};
::ArrowSchema upstream_c_schema;
ASSERT_TRUE(arrow::ExportSchema(*upstream_schema, &upstream_c_schema).ok());
ASSERT_OK_AND_ASSIGN(auto upstream_catalog,
Catalog::Create(dir_->Str(), {{Options::FILE_SYSTEM, "local"}}));
ASSERT_OK(upstream_catalog->CreateDatabase(upstream_db_name, {}, /*ignore_if_exists=*/true));
ASSERT_OK(upstream_catalog->CreateTable(
Identifier(upstream_db_name, upstream_table_name), &upstream_c_schema,
/*partition_keys=*/{}, /*primary_keys=*/{}, upstream_options,
/*ignore_if_exists=*/false));
std::string upstream_table_path =
PathUtil::JoinPath(dir_->Str(), upstream_db_name + ".db/" + upstream_table_name);
// Write 4 rows of b0/b1 data into the upstream table.
std::string upstream_raw_json = R"([
[0, "b0_data_0", "b1_data_0"],
[1, "b0_data_1", "b1_data_1"],
[2, "b0_data_2", "b1_data_2"],
[3, "b0_data_3", "b1_data_3"]
])";
auto upstream_raw_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(upstream_fields),
upstream_raw_json)
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto upstream_desc_array,
ConvertRawBlobToDescriptor(upstream_raw_array, {"b0", "b1"}));
ASSERT_OK_AND_ASSIGN(
auto upstream_commit_msgs,
WriteArray(upstream_table_path, {}, upstream_schema->field_names(), {upstream_desc_array}));
ASSERT_OK(Commit(upstream_table_path, upstream_commit_msgs));
// Create the downstream blob-view table that references both b0 and b1 of the upstream table.
arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
BlobUtils::ToArrowField("view", true)};
std::map<std::string, std::string> options = {
{Options::FILE_FORMAT, file_format}, {Options::BUCKET, "-1"},
{Options::ROW_TRACKING_ENABLED, "true"}, {Options::DATA_EVOLUTION_ENABLED, "true"},
{Options::BLOB_VIEW_FIELD, "view"}, {Options::FILE_SYSTEM, "local"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
// Build the view column mixing references to b0 (field_id=1) and b1 (field_id=2).
// - row 0: b0 row 0 -> "b0_data_0"
// - row 1: b1 row 1 -> "b1_data_1"
// - row 2: b0 row 2 -> "b0_data_2"
// - row 3: b1 row 3 -> "b1_data_3"
Identifier upstream_identifier(upstream_db_name, upstream_table_name);
auto append_view = [&](int32_t field_id, int64_t row_id, arrow::LargeBinaryBuilder* builder) {
BlobViewStruct view_struct(upstream_identifier, field_id, row_id);
ASSERT_OK_AND_ASSIGN(PAIMON_UNIQUE_PTR<Bytes> serialized, view_struct.Serialize(pool_));
ASSERT_TRUE(
builder
->Append(reinterpret_cast<const uint8_t*>(serialized->data()), serialized->size())
.ok());
};
arrow::LargeBinaryBuilder view_builder;
append_view(/*field_id=*/1, /*row_id=*/0, &view_builder);
append_view(/*field_id=*/2, /*row_id=*/1, &view_builder);
append_view(/*field_id=*/1, /*row_id=*/2, &view_builder);
append_view(/*field_id=*/2, /*row_id=*/3, &view_builder);
std::shared_ptr<arrow::Array> write_view_array;
ASSERT_TRUE(view_builder.Finish(&write_view_array).ok());
auto write_f0_array =
arrow::ipc::internal::json::ArrayFromJSON(arrow::int32(), R"([100,101,102,103])")
.ValueOrDie();
auto write_struct = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::StructArray::Make(arrow::ArrayVector({write_f0_array, write_view_array}),
std::vector<std::string>({"f0", "view"}))
.ValueOrDie());
auto schema = arrow::schema(fields);
ASSERT_OK_AND_ASSIGN(auto commit_msgs,
WriteArray(table_path, {}, schema->field_names(), {write_struct}));
ASSERT_OK(Commit(table_path, commit_msgs));
// Scan/read the downstream table and verify the resolved view blobs.
ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
ASSERT_OK_AND_ASSIGN(auto result,
ReadTable(table_path, schema->field_names(), plan, /*predicate=*/nullptr));
ASSERT_TRUE(result);
auto read_concat = arrow::Concatenate(result->chunks()).ValueOrDie();
auto read_struct = std::dynamic_pointer_cast<arrow::StructArray>(read_concat);
ASSERT_EQ(read_struct->length(), 4);
ASSERT_OK_AND_ASSIGN(auto result_array, ConvertDescriptorToRawBlob(read_struct, {"view"}));
std::string expected_json = R"([
[100, "b0_data_0"],
[101, "b1_data_1"],
[102, "b0_data_2"],
[103, "b1_data_3"]
])";
auto expected_struct = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), expected_json)
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(expected_struct));
ASSERT_TRUE(result_array->Equals(expected_with_rk))
<< "result_array:" << result_array->ToString() << std::endl
<< "expected:" << expected_with_rk->ToString();
}
TEST_P(BlobTableInteTest, TestBlobViewFieldWithMultipleUpstreamTables) {
auto file_format = GetParam();
if (file_format == "avro" || file_format == "mosaic") {
return;
}
// Upstream table 1: append_table_with_multi_blob, with two blob fields f5 (field_id=5) and
// f6 (field_id=6).
const std::string multi_blob_db_name = "append_table_with_multi_blob";
const std::string multi_blob_table_name = "append_table_with_multi_blob";
{
std::string src_db_path = paimon::test::GetDataDir() + file_format + "/" +
multi_blob_db_name + ".db/" + multi_blob_table_name;
std::string dst_db_path =
PathUtil::JoinPath(dir_->Str(), multi_blob_db_name + ".db/" + multi_blob_table_name);
ASSERT_TRUE(TestUtil::CopyDirectory(src_db_path, dst_db_path));
}
// Upstream table 2: blob_append_table_alter_table_with_cast_with_data_evolution, with one blob
// field blob (field_id=8).
const std::string alter_db_name = "blob_append_table_alter_table_with_cast_with_data_evolution";
const std::string alter_table_name =
"blob_append_table_alter_table_with_cast_with_data_evolution";
{
std::string src_db_path = paimon::test::GetDataDir() + file_format + "/" + alter_db_name +
".db/" + alter_table_name;
std::string dst_db_path =
PathUtil::JoinPath(dir_->Str(), alter_db_name + ".db/" + alter_table_name);
ASSERT_TRUE(TestUtil::CopyDirectory(src_db_path, dst_db_path));
}
arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
BlobUtils::ToArrowField("view1", true),
BlobUtils::ToArrowField("view2", true)};
std::map<std::string, std::string> options = {
{Options::FILE_FORMAT, file_format}, {Options::BUCKET, "-1"},
{Options::ROW_TRACKING_ENABLED, "true"}, {Options::DATA_EVOLUTION_ENABLED, "true"},
{Options::BLOB_VIEW_FIELD, "view1,view2"}, {Options::FILE_SYSTEM, "local"}};
CreateTable(fields, /*partition_keys=*/{}, options);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
Identifier multi_blob_identifier(multi_blob_db_name, multi_blob_table_name);
Identifier alter_identifier(alter_db_name, alter_table_name);
auto append_view = [&](const Identifier& identifier, int32_t field_id, int64_t row_id,
arrow::LargeBinaryBuilder* builder) {
BlobViewStruct view_struct(identifier, field_id, row_id);
ASSERT_OK_AND_ASSIGN(PAIMON_UNIQUE_PTR<Bytes> serialized, view_struct.Serialize(pool_));
ASSERT_TRUE(
builder
->Append(reinterpret_cast<const uint8_t*>(serialized->data()), serialized->size())
.ok());
};
// Build view1 column. References multi_blob.f5 (field_id=5) and f6 (field_id=6).
// Some upstream cells are referenced more than once on purpose.
// - row 0: f5 row 3 -> 'D' * 1024
// - row 1: f6 row 1 -> 'b' * 2048
// - row 2: f5 row 3 -> 'D' * 1024 (repeat of row 0)
// - row 3: f6 row 4 -> 'e' * 2048
// - row 4: f5 row 3 -> 'D' * 1024 (repeat of row 0)
// - row 5: f6 row 1 -> 'b' * 2048 (repeat of row 1)
// - row 6: f5 row 5 -> 'F' * 1024
// - row 7: f6 row 6 -> 'g' * 2048
arrow::LargeBinaryBuilder view1_builder;
append_view(multi_blob_identifier, /*field_id=*/5, /*row_id=*/3, &view1_builder);
append_view(multi_blob_identifier, /*field_id=*/6, /*row_id=*/1, &view1_builder);
append_view(multi_blob_identifier, /*field_id=*/5, /*row_id=*/3, &view1_builder);
append_view(multi_blob_identifier, /*field_id=*/6, /*row_id=*/4, &view1_builder);
append_view(multi_blob_identifier, /*field_id=*/5, /*row_id=*/3, &view1_builder);
append_view(multi_blob_identifier, /*field_id=*/6, /*row_id=*/1, &view1_builder);
append_view(multi_blob_identifier, /*field_id=*/5, /*row_id=*/5, &view1_builder);
append_view(multi_blob_identifier, /*field_id=*/6, /*row_id=*/6, &view1_builder);
std::shared_ptr<arrow::Array> write_view1_array;
ASSERT_TRUE(view1_builder.Finish(&write_view1_array).ok());
// Build view2 column. References alter.blob (field_id=8).
// Some upstream cells are referenced more than once on purpose.
// - row 0: blob row 0 -> "Lily"
// - row 1: blob row 5 -> "Apple"
// - row 2: blob row 0 -> "Lily" (repeat of row 0)
// - row 3: blob row 2 -> "Bob"
// - row 4: blob row 5 -> "Apple" (repeat of row 1)
// - row 5: blob row 9 -> "Elderberry"
// - row 6: blob row 0 -> "Lily" (repeat of row 0)
// - row 7: blob row 3 -> "Cindy"
arrow::LargeBinaryBuilder view2_builder;
append_view(alter_identifier, /*field_id=*/8, /*row_id=*/0, &view2_builder);
append_view(alter_identifier, /*field_id=*/8, /*row_id=*/5, &view2_builder);
append_view(alter_identifier, /*field_id=*/8, /*row_id=*/0, &view2_builder);
append_view(alter_identifier, /*field_id=*/8, /*row_id=*/2, &view2_builder);
append_view(alter_identifier, /*field_id=*/8, /*row_id=*/5, &view2_builder);
append_view(alter_identifier, /*field_id=*/8, /*row_id=*/9, &view2_builder);
append_view(alter_identifier, /*field_id=*/8, /*row_id=*/0, &view2_builder);
append_view(alter_identifier, /*field_id=*/8, /*row_id=*/3, &view2_builder);
std::shared_ptr<arrow::Array> write_view2_array;
ASSERT_TRUE(view2_builder.Finish(&write_view2_array).ok());
auto write_f0_array = arrow::ipc::internal::json::ArrayFromJSON(
arrow::int32(), R"([100,101,102,103,104,105,106,107])")
.ValueOrDie();
auto write_struct = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::StructArray::Make(
arrow::ArrayVector({write_f0_array, write_view1_array, write_view2_array}),
std::vector<std::string>({"f0", "view1", "view2"}))
.ValueOrDie());
// write & commit
auto schema = arrow::schema(fields);
ASSERT_OK_AND_ASSIGN(auto commit_msgs,
WriteArray(table_path, {}, schema->field_names(), {write_struct}));
ASSERT_OK(Commit(table_path, commit_msgs));
// Expected blob contents per referenced upstream cell.
std::string blob_f5_row3(1024, 'D'); // multi_blob.f5 row 3
std::string blob_f5_row5(1024, 'F'); // multi_blob.f5 row 5
std::string blob_f6_row1(2048, 'b'); // multi_blob.f6 row 1
std::string blob_f6_row4(2048, 'e'); // multi_blob.f6 row 4
std::string blob_f6_row6(2048, 'g'); // multi_blob.f6 row 6
std::vector<std::string> read_fields = {"view2", "view1", "f0"};
ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
ASSERT_OK_AND_ASSIGN(auto result,
ReadTable(table_path, read_fields, plan, /*predicate=*/nullptr));
ASSERT_TRUE(result);
auto read_concat = arrow::Concatenate(result->chunks()).ValueOrDie();
auto read_struct = std::dynamic_pointer_cast<arrow::StructArray>(read_concat);
ASSERT_EQ(read_struct->length(), 8);
ASSERT_OK_AND_ASSIGN(auto result_array,
ConvertDescriptorToRawBlob(read_struct, {"view1", "view2"}));
// Expected struct follows the requested (shuffled) column order: view2, view1, f0.
arrow::FieldVector expected_fields = {BlobUtils::ToArrowField("view2", true),
BlobUtils::ToArrowField("view1", true),
arrow::field("f0", arrow::int32())};
// clang-format off
std::string expected_json = R"([
["Lily", ")" + blob_f5_row3 + R"(", 100],
["Apple", ")" + blob_f6_row1 + R"(", 101],
["Lily", ")" + blob_f5_row3 + R"(", 102],
["Bob", ")" + blob_f6_row4 + R"(", 103],
["Apple", ")" + blob_f5_row3 + R"(", 104],
["Elderberry", ")" + blob_f6_row1 + R"(", 105],
["Lily", ")" + blob_f5_row5 + R"(", 106],
["Cindy", ")" + blob_f6_row6 + R"(", 107]
])";
// clang-format on
auto expected_struct = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(expected_fields), expected_json)
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(expected_struct));
ASSERT_TRUE(result_array->Equals(expected_with_rk))
<< "result_array:" << result_array->ToString() << std::endl
<< "expected:" << expected_with_rk->ToString();
}
TEST_P(BlobTableInteTest, TestBlobViewFailsWhenUpstreamTableAbsent) {
if (GetParam() == "mosaic") {
return;
}
arrow::FieldVector fields = CreateBlobViewTable(/*deletion_vectors_enabled=*/false);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields);
ASSERT_OK(WriteDanglingBlobViewRows(table_path));
// reading fails because the upstream table does not exist at the catalog-issued location
ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
ASSERT_NOK_WITH_MSG(ReadTable(table_path, schema->field_names(), plan, /*predicate=*/nullptr),
"not found latest schema");
}
TEST_P(BlobTableInteTest, TestBlobViewSkipsDanglingReferenceOfDeletedRow) {
if (GetParam() == "mosaic") {
return;
}
arrow::FieldVector fields = CreateBlobViewTable(/*deletion_vectors_enabled=*/true);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields);
// resolving row 0 would fail the whole query, so the pre-read that collects the structs to
// resolve must honor the deletion vector too
ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> commit_msgs,
WriteDanglingBlobViewRows(table_path, /*first_row_id=*/0));
// without the deletion vector the dangling reference fails the read
ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
ASSERT_NOK_WITH_MSG(ReadTable(table_path, schema->field_names(), plan, /*predicate=*/nullptr),
"not found latest schema");
ASSERT_OK_AND_ASSIGN(std::string anchor_file_name,
DeletionVectorTestHelper::RetrieveAnchorFileName(commit_msgs));
ASSERT_OK(CommitDeletionVectors(table_path, commit_msgs[0],
{{anchor_file_name, /*deleted_positions=*/{0}}}));
// the deleted row is never resolved, so only the surviving row is returned
ASSERT_OK_AND_ASSIGN(auto dv_plan, ScanTable(table_path));
ASSERT_OK_AND_ASSIGN(auto result, ReadTable(table_path, schema->field_names(), dv_plan,
/*predicate=*/nullptr));
ASSERT_TRUE(result);
auto read_concat = arrow::Concatenate(result->chunks()).ValueOrDie();
auto read_struct = std::dynamic_pointer_cast<arrow::StructArray>(read_concat);
ASSERT_EQ(read_struct->length(), 1);
auto expected_struct = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([[101, null]])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(expected_struct));
ASSERT_TRUE(read_struct->Equals(expected_with_rk))
<< "read_struct:" << read_struct->ToString() << std::endl
<< "expected:" << expected_with_rk->ToString();
}
TEST_P(BlobTableInteTest, TestBlobViewSkipsDanglingReferenceInEveryRowRangeGroup) {
if (GetParam() == "mosaic") {
return;
}
arrow::FieldVector fields = CreateBlobViewTable(/*deletion_vectors_enabled=*/true);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields);
// two commits, so the two pairs of rows form two row range groups. The blob view pre-read
// walks the groups one by one; a group it drops, or reads with another group's deletion
// vector, still hands its dangling reference to the resolver and fails the whole read.
ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> commit_msgs0,
WriteDanglingBlobViewRows(table_path, /*first_row_id=*/0));
ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> commit_msgs1,
WriteDanglingBlobViewRows(table_path, /*first_row_id=*/2));
// without deletion vectors both groups' dangling references fail the read
ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
ASSERT_NOK_WITH_MSG(ReadTable(table_path, schema->field_names(), plan, /*predicate=*/nullptr),
"not found latest schema");
// each group's anchor gets a deletion vector dropping its row 0, the dangling one
ASSERT_OK_AND_ASSIGN(std::string anchor_file_name0,
DeletionVectorTestHelper::RetrieveAnchorFileName(commit_msgs0));
ASSERT_OK_AND_ASSIGN(std::string anchor_file_name1,
DeletionVectorTestHelper::RetrieveAnchorFileName(commit_msgs1));
ASSERT_NE(anchor_file_name0, anchor_file_name1);
ASSERT_OK(CommitDeletionVectors(table_path, commit_msgs0[0],
{{anchor_file_name0, /*deleted_positions=*/{0}},
{anchor_file_name1, /*deleted_positions=*/{0}}}));
// both groups keep only their row 1, which holds no blob view at all
auto expected_struct = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields),
R"([[101, null], [101, null]])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_struct));
}
// Pins one of the limitations documented on CreateBlobViewReader: the pre-read takes no
// predicate, so filtering a row out that way does not stop its blob view reference from being
// resolved. This asserts the current behavior, not a desired one.
TEST_P(BlobTableInteTest, TestBlobViewPreReadHonorsRowRangesNotPredicate) {
if (GetParam() == "mosaic") {
return;
}
arrow::FieldVector fields = CreateBlobViewTable(/*deletion_vectors_enabled=*/false);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields);
// row 0 (f0 = 100) holds the dangling reference, row 1 (f0 = 101) holds no blob view
ASSERT_OK(WriteDanglingBlobViewRows(table_path));
// reading the whole split fails on the dangling reference of row 0
ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
ASSERT_NOK_WITH_MSG(ReadTable(table_path, schema->field_names(), plan, /*predicate=*/nullptr),
"not found latest schema");
// a predicate keeping only row 1 does not rescue the query: it is evaluated above the split
// read, while the pre-read takes none, so row 0's reference is still collected and resolved
auto keep_row_1 = PredicateBuilder::Equal(/*field_index=*/0, /*field_name=*/"f0",
FieldType::INT, Literal(101));
ASSERT_OK_AND_ASSIGN(auto filtered_plan, ScanTable(table_path, keep_row_1));
ASSERT_NOK_WITH_MSG(ReadTable(table_path, schema->field_names(), filtered_plan, keep_row_1),
"not found latest schema");
// dropping the same row through the row ranges does: the pre-read honors them the same way
// the main read does, so row 0 is never resolved
auto expected_struct = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([[101, null]])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_struct,
/*predicate=*/nullptr, /*row_ranges=*/{Range(1, 1)}));
}
TEST_P(BlobTableInteTest, TestReadBlobDescriptorFieldFromJava) {
auto file_format = GetParam();
if (file_format == "avro" || file_format == "mosaic") {
return;
}
std::string table_path =
GetDataDir() + "/" + file_format + "/blob_desc_field.db/blob_desc_field";
arrow::FieldVector fields = {
arrow::field("f0", arrow::int32()), BlobUtils::ToArrowField("b0", true),
BlobUtils::ToArrowField("b1", true), BlobUtils::ToArrowField("b2", true),
BlobUtils::ToArrowField("b3", true)};
auto schema = arrow::schema(fields);
// b0: all non-null, b1: has nulls, b2: all non-null, b3: has nulls
std::string raw_json = R"([
[1, "img_0", null, "raw_2_0", "raw_3_0"],
[2, "img_1", "vid_1", "raw_2_1", null ],
[3, "img_2", null, "raw_2_2", "raw_3_2" ]
])";
auto raw_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), raw_json).ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
std::map<std::string, std::string> read_options = {{Options::BLOB_AS_DESCRIPTOR, "false"}};
ASSERT_OK_AND_ASSIGN(auto result, ReadTable(table_path, schema->field_names(), plan,
/*predicate=*/nullptr, read_options));
ASSERT_TRUE(result);
auto read_concat = arrow::Concatenate(result->chunks()).ValueOrDie();
auto read_struct = std::dynamic_pointer_cast<arrow::StructArray>(read_concat);
// After read, b0 and b1 are both descriptor-stored; resolve all back to raw bytes.
// Java-generated descriptors may contain absolute paths from the generation machine.
// Rewrite them to the portable blob directories inside the copied table path.
BlobDescriptorPathRewrite rewrite{table_path, {"raw_blob", "external_blob"}};
ASSERT_OK_AND_ASSIGN(auto resolved,
ConvertDescriptorToRawBlob(read_struct, {"b0", "b1"}, rewrite));
ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(raw_array));
ASSERT_TRUE(resolved->Equals(expected_with_rk));
}
TEST_P(BlobTableInteTest, TestReadArrayBlobTableFromJava) {
if (GetParam() != "parquet") {
GTEST_SKIP() << "the Java fixture uses Parquet";
}
const std::string table_path = GetDataDir() + "/parquet/array_blob_java.db/array_blob_java";
const std::vector<std::string> read_fields = {"id", "array_payloads"};
for (int64_t snapshot_id : {1, 2, 3}) {
ScanContextBuilder scan_builder(table_path);
scan_builder.AddOption(Options::SCAN_SNAPSHOT_ID, std::to_string(snapshot_id));
ASSERT_OK_AND_ASSIGN(auto scan_context, scan_builder.Finish());
ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context)));
ASSERT_OK_AND_ASSIGN(auto plan, table_scan->CreatePlan());
size_t array_layer_count = 0;
for (const auto& split : plan->Splits()) {
auto data_split = std::dynamic_pointer_cast<DataSplitImpl>(split);
ASSERT_TRUE(data_split);
for (const auto& file : data_split->DataFiles()) {
if (file->write_cols ==
std::optional<std::vector<std::string>>({"array_payloads"})) {
++array_layer_count;
}
}
}
ASSERT_EQ(snapshot_id, static_cast<int64_t>(array_layer_count));
for (bool blob_as_descriptor : {false, true}) {
std::map<std::string, std::string> read_options = {
{Options::BLOB_AS_DESCRIPTOR, blob_as_descriptor ? "true" : "false"}};
ASSERT_OK_AND_ASSIGN(auto result, ReadTable(table_path, read_fields, plan,
/*predicate=*/nullptr, read_options));
ASSERT_TRUE(result);
auto combined = arrow::Concatenate(result->chunks()).ValueOrDie();
auto rows = std::dynamic_pointer_cast<arrow::StructArray>(combined);
ASSERT_TRUE(rows);
ASSERT_EQ(4, rows->length());
const auto& ids = checked_cast<const arrow::Int32Array&>(*rows->GetFieldByName("id"));
for (int64_t i = 0; i < ids.length(); ++i) {
ASSERT_EQ(i + 1, ids.Value(i));
}
const std::string expected_json =
snapshot_id == 1
? R"json([["array-alpha", null, "", "array-omega"], null, [], ["array-single"]])json"
: R"json([["array-alpha", null, "", "array-omega"], ["array-updated"], [], ["array-single"]])json";
CheckArrayBlobColumn(rows, "array_payloads", expected_json, blob_as_descriptor);
}
}
}
TEST_P(BlobTableInteTest, TestReadMapBlobTableFromJava) {
if (GetParam() != "parquet") {
GTEST_SKIP() << "the Java fixture uses Parquet";
}
const std::string table_path = GetDataDir() + "/parquet/map_blob_java.db/map_blob_java";
const std::vector<std::string> read_fields = {"id",
"string_payloads",
"boolean_payloads",
"tinyint_payloads",
"smallint_payloads",
"int_payloads",
"bigint_payloads",
"date_payloads",
"binary_payloads",
"compact_decimal_payloads",
"large_decimal_payloads"};
for (int64_t snapshot_id : {1, 3}) {
ScanContextBuilder scan_builder(table_path);
scan_builder.AddOption(Options::SCAN_SNAPSHOT_ID, std::to_string(snapshot_id));
ASSERT_OK_AND_ASSIGN(auto scan_context, scan_builder.Finish());
ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context)));
ASSERT_OK_AND_ASSIGN(auto plan, table_scan->CreatePlan());
size_t string_layer_count = 0;
for (const auto& split : plan->Splits()) {
auto data_split = std::dynamic_pointer_cast<DataSplitImpl>(split);
ASSERT_TRUE(data_split);
for (const auto& file : data_split->DataFiles()) {
if (file->write_cols ==
std::optional<std::vector<std::string>>({"string_payloads"})) {
++string_layer_count;
}
}
}
ASSERT_EQ(snapshot_id == 1 ? 1 : 3, string_layer_count);
for (bool blob_as_descriptor : {false, true}) {
std::map<std::string, std::string> read_options = {
{Options::BLOB_AS_DESCRIPTOR, blob_as_descriptor ? "true" : "false"}};
ASSERT_OK_AND_ASSIGN(auto result, ReadTable(table_path, read_fields, plan,
/*predicate=*/nullptr, read_options));
ASSERT_TRUE(result);
auto combined = arrow::Concatenate(result->chunks()).ValueOrDie();
auto rows = std::dynamic_pointer_cast<arrow::StructArray>(combined);
ASSERT_TRUE(rows);
ASSERT_EQ(4, rows->length());
const auto& ids = checked_cast<const arrow::Int32Array&>(*rows->GetFieldByName("id"));
for (int64_t i = 0; i < ids.length(); ++i) {
ASSERT_EQ(i + 1, ids.Value(i));
}
CheckMapBlobColumn(
rows, "string_payloads",
R"json([[["", "string-empty"], ["alpha", "string-alpha"]], [], null, [["omega", "string-omega"]]])json",
blob_as_descriptor);
CheckMapBlobColumn(
rows, "boolean_payloads",
R"json([[[false, "bool-false"], [true, "bool-true"]], null, null, null])json",
blob_as_descriptor);
CheckMapBlobColumn(
rows, "tinyint_payloads",
R"json([[[-128, "tiny-min"], [-1, "tiny-negative"], [127, "tiny-max"]], null, null, null])json",
blob_as_descriptor);
CheckMapBlobColumn(
rows, "smallint_payloads",
R"json([[[-32768, "small-min"], [-1, "small-negative"], [32767, "small-max"]], null, null, null])json",
blob_as_descriptor);
CheckMapBlobColumn(
rows, "int_payloads",
R"json([[[-2147483648, "int-min"], [-1, "int-negative"], [2147483647, "int-max"]], null, null, null])json",
blob_as_descriptor);
CheckMapBlobColumn(
rows, "bigint_payloads",
R"json([[[-9223372036854775808, "big-min"], [-1, "big-negative"], [9223372036854775807, "big-max"]], null, null, null])json",
blob_as_descriptor);
CheckMapBlobColumn(
rows, "date_payloads",
R"json([[[-1, "date-negative"], [0, "date-epoch"]], null, null, null])json",
blob_as_descriptor);
CheckMapBlobColumn(
rows, "compact_decimal_payloads",
R"json([[["-99999999.99", "compact-negative"], ["99999999.99", "compact-positive"]], null, null, null])json",
blob_as_descriptor);
CheckMapBlobColumn(
rows, "large_decimal_payloads",
R"json([[["-999999999999999999.99", "large-negative"], ["999999999999999999.99", "large-positive"]], null, null, null])json",
blob_as_descriptor);
auto binary_map =
std::dynamic_pointer_cast<arrow::MapArray>(rows->GetFieldByName("binary_payloads"));
ASSERT_TRUE(binary_map);
ASSERT_OK_AND_ASSIGN(auto normalized_binary,
NormalizeMapBlobValues(binary_map, blob_as_descriptor));
const auto& binary_keys =
checked_cast<const arrow::BinaryArray&>(*normalized_binary->keys());
const auto& binary_values =
checked_cast<const arrow::LargeBinaryArray&>(*normalized_binary->items());
ASSERT_EQ(2, normalized_binary->value_length(0));
ASSERT_EQ("", binary_keys.GetString(0));
ASSERT_EQ(std::string("\0\xff\1\2", 4), binary_keys.GetString(1));
ASSERT_EQ("binary-empty", binary_values.GetString(0));
ASSERT_EQ("binary-bytes", binary_values.GetString(1));
}
}
}
} // namespace paimon::test