Copilot commented on code in PR #123: URL: https://github.com/apache/paimon-cpp/pull/123#discussion_r3472401125
########## src/paimon/testing/utils/read_result_collector.h: ########## @@ -0,0 +1,171 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include <unistd.h> + +#include <memory> +#include <string> +#include <utility> +#include <vector> + +#include "arrow/api.h" +#include "arrow/c/bridge.h" +#include "arrow/compute/api.h" +#include "paimon/common/reader/reader_utils.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/common/utils/date_time_utils.h" +#include "paimon/core/io/key_value_data_file_record_reader.h" +#include "paimon/core/key_value.h" +#include "paimon/reader/batch_reader.h" +#include "paimon/testing/utils/dict_array_converter.h" +namespace paimon::test { +class ReadResultCollector { + public: + ReadResultCollector() = delete; + ~ReadResultCollector() = delete; + + template <typename ReaderType, typename IteratorType> + static Result<std::vector<KeyValue>> CollectKeyValueResult(ReaderType* reader) { + std::vector<KeyValue> results; + while (true) { + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<IteratorType> iterator, reader->NextBatch()); + if (iterator == nullptr) { + break; + } + while (true) { + if constexpr (std::is_same_v<IteratorType, KeyValueRecordReader::Iterator>) { + PAIMON_ASSIGN_OR_RAISE(bool has_next, iterator->HasNext()); + if (!has_next) { + break; + } + PAIMON_ASSIGN_OR_RAISE(KeyValue kv, iterator->Next()); + results.emplace_back(std::move(kv)); + } else { + PAIMON_ASSIGN_OR_RAISE(bool has_next, iterator->HasNext()); + if (!has_next) { + break; + } + KeyValue kv = iterator->Next(); + results.emplace_back(std::move(kv)); + } + } + } + return results; + } + + static Result<std::shared_ptr<arrow::ChunkedArray>> CollectResult(BatchReader* batch_reader) { + return CollectResult(batch_reader, /*max simulated data processing time*/ 0); + } + + // will convert dictionary array to string array for comparing results + static Result<std::shared_ptr<arrow::ChunkedArray>> CollectResult( + BatchReader* batch_reader, int64_t max_data_processing_time_in_us) { + arrow::ArrayVector result_array_vector; + int64_t seed = DateTimeUtils::GetCurrentUTCTimeUs(); + std::srand(seed); + while (true) { + // Prioritize calling NextBatch. If it fails (paimon inner reader e.g., + // PrefetchBatchReader, ApplyBitmapIndexBatchReader...), call NextBatchWithBitmap. + auto batch_result = batch_reader->NextBatch(); + BatchReader::ReadBatch batch; + if (!batch_result.ok()) { + if (batch_result.status().ToString().find("should use NextBatchWithBitmap") != + std::string::npos) { + PAIMON_ASSIGN_OR_RAISE(BatchReader::ReadBatchWithBitmap batch_with_bitmap, + batch_reader->NextBatchWithBitmap()); + if (BatchReader::IsEofBatch(batch_with_bitmap)) { + break; + } + assert(!batch_with_bitmap.second.IsEmpty()); + PAIMON_ASSIGN_OR_RAISE( + batch, ReaderUtils::ApplyBitmapToReadBatch(std::move(batch_with_bitmap), + arrow::default_memory_pool())); + } else { + return batch_result.status(); + } + } else { + batch = std::move(batch_result).value(); + if (BatchReader::IsEofBatch(batch)) { + break; + } + } + auto& [c_array, c_schema] = batch; + assert(c_array->length > 0); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(auto result_array, + arrow::ImportArray(c_array.get(), c_schema.get())); + result_array_vector.push_back(result_array); + if (max_data_processing_time_in_us > 0) { + usleep(std::rand() % max_data_processing_time_in_us); + } + } + if (result_array_vector.empty()) { + return std::shared_ptr<arrow::ChunkedArray>(); + } Review Comment: `CollectResult()` returns a null `shared_ptr<arrow::ChunkedArray>` when no batches are read. Callers (e.g., `TestHelper::ReadAndCheckResult*`) dereference the returned pointer unconditionally, which can cause a crash for empty results. Returning an explicit error makes the failure mode deterministic and easier to diagnose. ########## src/paimon/testing/utils/test_helper.h: ########## @@ -0,0 +1,443 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include <map> +#include <memory> +#include <string> +#include <utility> +#include <vector> + +#include "arrow/c/bridge.h" +#include "arrow/ipc/api.h" +#include "paimon/api.h" +#include "paimon/catalog/catalog.h" +#include "paimon/commit_context.h" +#include "paimon/common/data/blob_descriptor.h" +#include "paimon/common/data/blob_utils.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/common/utils/path_util.h" Review Comment: `test_helper.h` includes `paimon/api.h` (which is not present in this repo) and also uses a number of types/macros (e.g., `Options`, `ScopeGuard`, `StringUtils`, `std::optional`, `std::cout`, `std::stable_sort`, `assert`) without including the headers that define them. This is likely to break compilation depending on include order. ########## src/paimon/testing/utils/test_helper.h: ########## @@ -0,0 +1,443 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include <map> +#include <memory> +#include <string> +#include <utility> +#include <vector> + +#include "arrow/c/bridge.h" +#include "arrow/ipc/api.h" +#include "paimon/api.h" +#include "paimon/catalog/catalog.h" +#include "paimon/commit_context.h" +#include "paimon/common/data/blob_descriptor.h" +#include "paimon/common/data/blob_utils.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/common/utils/path_util.h" +#include "paimon/core/operation/append_only_file_store_write.h" +#include "paimon/core/operation/file_store_commit_impl.h" +#include "paimon/core/schema/schema_manager.h" +#include "paimon/core/table/sink/commit_message_impl.h" +#include "paimon/core/utils/file_store_path_factory.h" +#include "paimon/core/utils/snapshot_manager.h" +#include "paimon/data/blob.h" +#include "paimon/file_store_commit.h" +#include "paimon/file_store_write.h" +#include "paimon/fs/file_system_factory.h" +#include "paimon/record_batch.h" +#include "paimon/result.h" +#include "paimon/table/source/startup_mode.h" +#include "paimon/table/source/table_scan.h" +#include "paimon/testing/utils/read_result_collector.h" +#include "paimon/testing/utils/testharness.h" +#include "paimon/write_context.h" +namespace paimon::test { + +class TestHelper { + public: + static Result<std::unique_ptr<TestHelper>> Create( + const std::string& root_path, const std::shared_ptr<arrow::Schema>& schema, + const std::vector<std::string>& partition_keys, + const std::vector<std::string>& primary_keys, + const std::map<std::string, std::string>& options, bool is_streaming_mode, + bool ignore_if_exists = false) { + // only for test && only check the key + auto new_options = options; + new_options["enable-object-store-catalog-in-inte-test"] = ""; + PAIMON_ASSIGN_OR_RAISE(auto catalog, Catalog::Create(root_path, new_options)); + PAIMON_RETURN_NOT_OK(catalog->CreateDatabase("foo", new_options, ignore_if_exists)); + ::ArrowSchema c_schema; + ScopeGuard guard([schema = &c_schema]() { ArrowSchemaRelease(schema); }); + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*schema, &c_schema)); + PAIMON_RETURN_NOT_OK(catalog->CreateTable(Identifier("foo", "bar"), &c_schema, + partition_keys, primary_keys, new_options, + ignore_if_exists)); + std::string table_path = PathUtil::JoinPath(root_path, "foo.db/bar"); + return Create(table_path, new_options, is_streaming_mode); + } + + static Result<std::unique_ptr<TestHelper>> Create( + const std::string& table_path, const std::map<std::string, std::string>& options, + bool is_streaming_mode) { + std::string file_system_identifier = "local"; + auto fs_iter = options.find(Options::FILE_SYSTEM); + if (fs_iter != options.end()) { + file_system_identifier = StringUtils::ToLowerCase(fs_iter->second); + } + PAIMON_ASSIGN_OR_RAISE(auto file_system, + FileSystemFactory::Get(file_system_identifier, table_path, options)); + std::string commit_user = "commit_user"; + WriteContextBuilder context_builder(table_path, commit_user); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<WriteContext> write_context, + context_builder.SetOptions(options) + .WithStreamingMode(is_streaming_mode) + .WithWriteId(12345) + .Finish()); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileStoreWrite> write, + FileStoreWrite::Create(std::move(write_context))); + std::map<std::string, std::string> new_options = options; + // only for test && only check the key + new_options["enable-pk-commit-in-inte-test"] = ""; + new_options["enable-object-store-commit-in-inte-test"] = ""; + CommitContextBuilder commit_context_builder(table_path, commit_user); + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr<CommitContext> commit_context, + commit_context_builder.SetOptions(new_options).IgnoreEmptyCommit(false).Finish()); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileStoreCommit> commit, + FileStoreCommit::Create(std::move(commit_context))); + return std::unique_ptr<TestHelper>(new TestHelper(std::move(file_system), std::move(write), + std::move(commit), commit_user, + table_path, options)); + } + + static Result<std::unique_ptr<RecordBatch>> MakeRecordBatch( + const std::shared_ptr<arrow::DataType>& data_type, const std::string& data_str, + const std::map<std::string, std::string>& partition_map, int32_t bucket, + const std::vector<RecordBatch::RowKind>& row_kinds) { + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + auto array, arrow::ipc::internal::json::ArrayFromJSON(data_type, data_str)); + ::ArrowArray arrow_array; + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*array, &arrow_array)); + RecordBatchBuilder batch_builder(&arrow_array); + return batch_builder.SetPartition(partition_map) + .SetBucket(bucket) + .SetRowKinds(row_kinds) + .Finish(); + } + + Result<std::vector<std::shared_ptr<CommitMessage>>> WriteAndCommit( + std::unique_ptr<RecordBatch>&& record_batch, int64_t commit_identifier, + const std::optional<std::vector<std::shared_ptr<CommitMessage>>>& + expected_commit_messages) { + std::vector<std::unique_ptr<RecordBatch>> batches; + batches.emplace_back(std::move(record_batch)); + return WriteAndCommit(std::move(batches), commit_identifier, expected_commit_messages); + } + + Result<std::vector<std::shared_ptr<CommitMessage>>> WriteAndCommit( + std::vector<std::unique_ptr<RecordBatch>>&& record_batches, int64_t commit_identifier, + const std::optional<std::vector<std::shared_ptr<CommitMessage>>>& + expected_commit_messages) { + for (auto& record_batch : record_batches) { + PAIMON_RETURN_NOT_OK(write_->Write(std::move(record_batch))); + } + PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<CommitMessage>> commit_messages, + write_->PrepareCommit(/*wait_compaction=*/false, commit_identifier)); + if (expected_commit_messages) { + CheckCommitMessages(expected_commit_messages.value(), commit_messages); + CheckExternalPath(commit_messages); + } + PAIMON_RETURN_NOT_OK(commit_->Commit(commit_messages, commit_identifier)); + return commit_messages; + } + + Result<std::vector<std::shared_ptr<Split>>> NewScan(StartupMode startup_mode, + std::optional<int64_t> snapshot_id, + bool is_streaming = true) { + ScanContextBuilder scan_context_builder(table_path_); + scan_context_builder.WithStreamingMode(is_streaming) + .SetOptions(options_) + .AddOption(Options::SCAN_MODE, startup_mode.ToString()); + if (snapshot_id) { + scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, + std::to_string(snapshot_id.value())); + } + PAIMON_ASSIGN_OR_RAISE(auto scan_context, scan_context_builder.Finish()); + PAIMON_ASSIGN_OR_RAISE(scan_, TableScan::Create(std::move(scan_context))); + return Scan(); + } + + Result<std::vector<std::shared_ptr<Split>>> Scan() { + if (scan_ == nullptr) { + return Status::Invalid("need call NewScan first"); + } + PAIMON_ASSIGN_OR_RAISE(auto result_plan, scan_->CreatePlan()); + return result_plan->Splits(); + } + + static Result<bool> CheckBlobsEqual(const std::vector<std::shared_ptr<Blob>>& result_blobs, + const std::vector<std::shared_ptr<Blob>>& expected_blobs, + const std::shared_ptr<FileSystem>& fs) { + if (result_blobs.size() != expected_blobs.size()) { + std::cout << "[result_blobs.size]: " << result_blobs.size() << std::endl; + std::cout << "[expected_blobs.size]: " << expected_blobs.size() << std::endl; + return false; + } + for (uint32_t i = 0; i < result_blobs.size(); ++i) { + PAIMON_ASSIGN_OR_RAISE(auto result_stream, result_blobs[i]->NewInputStream(fs)); + PAIMON_ASSIGN_OR_RAISE(auto expected_stream, expected_blobs[i]->NewInputStream(fs)); + + PAIMON_ASSIGN_OR_RAISE(uint64_t result_length, result_stream->Length()); + PAIMON_ASSIGN_OR_RAISE(uint64_t expected_length, expected_stream->Length()); + if (result_length != expected_length) { + auto result_descriptor_bytes = result_blobs[i]->ToDescriptor(GetDefaultPool()); + auto expected_descriptor_bytes = expected_blobs[i]->ToDescriptor(GetDefaultPool()); + PAIMON_ASSIGN_OR_RAISE( + auto result_descriptor, + BlobDescriptor::Deserialize(result_descriptor_bytes->data(), + result_descriptor_bytes->size())); + PAIMON_ASSIGN_OR_RAISE( + auto expected_descriptor, + BlobDescriptor::Deserialize(expected_descriptor_bytes->data(), + expected_descriptor_bytes->size())); + std::cout << "blobs[" << i << "]: " << std::endl; + std::cout << "[result_length(" << result_length << ") != expected_length(" + << expected_length << ")]" << std::endl; + std::cout << "[result_descriptor]: " << result_descriptor->ToString() << std::endl; + std::cout << "[expected_descriptor]: " << expected_descriptor->ToString() + << std::endl; + return false; + } + + std::vector<char> result_bytes(result_length, 0); + std::vector<char> expected_bytes(expected_length, 0); + PAIMON_RETURN_NOT_OK(result_stream->Read(result_bytes.data(), result_length)); + PAIMON_RETURN_NOT_OK(expected_stream->Read(expected_bytes.data(), expected_length)); + if (result_bytes != expected_bytes) { + std::cout << "blobs[" << i << "]: " << std::endl; + std::cout << "[result_bytes != expected_bytes]" << std::endl; + return false; + } + } + return true; + } + + static Result<std::vector<std::shared_ptr<Blob>>> ToBlobs( + const std::shared_ptr<arrow::StructArray>& blob_struct_array) { + std::vector<std::shared_ptr<Blob>> result_blobs; + auto child_array = blob_struct_array->field(0); + assert(blob_struct_array->num_fields() == 1); + assert(blob_struct_array->null_count() == 0); + assert(child_array->null_count() == 0); + assert(child_array->type_id() == arrow::Type::type::LARGE_BINARY); + + const auto& blob_array = + arrow::internal::checked_cast<const arrow::LargeBinaryArray&>(*child_array); + for (int64_t i = 0; i < blob_array.length(); ++i) { + std::string_view descriptor = blob_array.GetView(i); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Blob> blob, + Blob::FromDescriptor(descriptor.data(), descriptor.size())); + result_blobs.emplace_back(blob); + } + return result_blobs; + } + + // need to reconstruct the blob array, because the array in read result do not have blob meta + Result<std::shared_ptr<arrow::Array>> ReconstructBlobArray( + const std::shared_ptr<arrow::Array>& array, const std::shared_ptr<arrow::Schema>& schema) { + ::ArrowArray c_array; + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*array, &c_array)); + ::ArrowSchema new_c_schema; + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*schema, &new_c_schema)); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(auto new_array, + arrow::ImportArray(&c_array, &new_c_schema)); + return new_array; + } + + Result<bool> ReadAndCheckResultForBlobTable( + const std::shared_ptr<arrow::Schema>& all_columns_schema, + const std::vector<std::shared_ptr<Split>>& splits, const std::string& main_expected_json, + const std::vector<PAIMON_UNIQUE_PTR<Bytes>>& expected_blob_descriptors) { + ReadContextBuilder read_context_builder(table_path_); + 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(batch_reader.get())); + + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(auto concat_array, + arrow::Concatenate(read_result->chunks())); + PAIMON_ASSIGN_OR_RAISE(auto reconstruct_array, + ReconstructBlobArray(concat_array, all_columns_schema)); + PAIMON_ASSIGN_OR_RAISE( + auto separated_array, + BlobUtils::SeparateBlobArray( + std::dynamic_pointer_cast<arrow::StructArray>(reconstruct_array))); + + arrow::EqualOptions equal_options = arrow::EqualOptions::Defaults(); + + // check main columns + auto separated_schema = BlobUtils::SeparateBlobSchema(all_columns_schema); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + auto main_expected_array, + arrow::ipc::internal::json::ArrayFromJSON( + arrow::struct_(separated_schema.main_schema->fields()), main_expected_json)); + auto main_expected_chunk_array = std::make_shared<arrow::ChunkedArray>(main_expected_array); + bool main_equal = main_expected_chunk_array->Equals( + arrow::ChunkedArray(separated_array.main_array), equal_options.diff_sink(&std::cout)); + if (!main_equal) { + std::cout << "[expected_data_type]" << main_expected_chunk_array->type()->ToString() + << std::endl; + std::cout << "[actual_data_type]" << separated_array.main_array->type()->ToString() + << std::endl; + std::cout << "[expected]:" << main_expected_chunk_array->ToString() << std::endl; + std::cout << "[actual]: " << separated_array.main_array->ToString() << std::endl; + } + + // check blob column + std::vector<std::shared_ptr<Blob>> expected_blobs; + for (const auto& descriptor : expected_blob_descriptors) { + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Blob> blob, + Blob::FromDescriptor(descriptor->data(), descriptor->size())); + expected_blobs.emplace_back(blob); + } + PAIMON_ASSIGN_OR_RAISE(auto result_blobs, ToBlobs(separated_array.blob_array)); + PAIMON_ASSIGN_OR_RAISE(bool blob_equal, CheckBlobsEqual(result_blobs, expected_blobs, fs_)); + + table_read.reset(); + return main_equal && blob_equal; + } + + Result<bool> ReadAndCheckResult(const std::shared_ptr<arrow::DataType>& data_type, + const std::vector<std::shared_ptr<Split>>& splits, + const std::string& expected_result) { + ReadContextBuilder read_context_builder(table_path_); + 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(batch_reader.get())); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + auto expected_array, + arrow::ipc::internal::json::ArrayFromJSON(data_type, expected_result)); + auto expected_chunk_array = std::make_shared<arrow::ChunkedArray>(expected_array); + + arrow::EqualOptions equal_options = arrow::EqualOptions::Defaults(); + bool is_equal = + expected_chunk_array->Equals(read_result, equal_options.diff_sink(&std::cout)); + + if (!is_equal) { + std::cout << "[expected_data_type]" << expected_chunk_array->type()->ToString() + << std::endl; + std::cout << "[actual_data_type]" << read_result->type()->ToString() << std::endl; + std::cout << "[expected]:" << expected_chunk_array->ToString() << std::endl; + std::cout << "[actual]: " << read_result->ToString() << std::endl; + } + table_read.reset(); + return is_equal; + } + + Result<std::optional<Snapshot>> LatestSnapshot() const { + auto commit_impl = dynamic_cast<FileStoreCommitImpl*>(commit_.get()); + return commit_impl->snapshot_manager_->LatestSnapshotOfUser(commit_user_); + } + + Result<std::optional<std::shared_ptr<TableSchema>>> LatestSchema() const { + auto commit_impl = dynamic_cast<FileStoreCommitImpl*>(commit_.get()); + return commit_impl->schema_manager_->Latest(); + } Review Comment: `LatestSchema()` downcasts to `FileStoreCommitImpl` and accesses `schema_manager_` directly, which bypasses the public API and is likely to break compilation (member access) and/or behavior if commit implementation changes. ########## src/paimon/testing/utils/test_helper.h: ########## @@ -0,0 +1,443 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include <map> +#include <memory> +#include <string> +#include <utility> +#include <vector> + +#include "arrow/c/bridge.h" +#include "arrow/ipc/api.h" +#include "paimon/api.h" +#include "paimon/catalog/catalog.h" +#include "paimon/commit_context.h" +#include "paimon/common/data/blob_descriptor.h" +#include "paimon/common/data/blob_utils.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/common/utils/path_util.h" +#include "paimon/core/operation/append_only_file_store_write.h" +#include "paimon/core/operation/file_store_commit_impl.h" +#include "paimon/core/schema/schema_manager.h" +#include "paimon/core/table/sink/commit_message_impl.h" +#include "paimon/core/utils/file_store_path_factory.h" +#include "paimon/core/utils/snapshot_manager.h" +#include "paimon/data/blob.h" +#include "paimon/file_store_commit.h" +#include "paimon/file_store_write.h" +#include "paimon/fs/file_system_factory.h" +#include "paimon/record_batch.h" +#include "paimon/result.h" +#include "paimon/table/source/startup_mode.h" +#include "paimon/table/source/table_scan.h" +#include "paimon/testing/utils/read_result_collector.h" +#include "paimon/testing/utils/testharness.h" +#include "paimon/write_context.h" +namespace paimon::test { + +class TestHelper { + public: + static Result<std::unique_ptr<TestHelper>> Create( + const std::string& root_path, const std::shared_ptr<arrow::Schema>& schema, + const std::vector<std::string>& partition_keys, + const std::vector<std::string>& primary_keys, + const std::map<std::string, std::string>& options, bool is_streaming_mode, + bool ignore_if_exists = false) { + // only for test && only check the key + auto new_options = options; + new_options["enable-object-store-catalog-in-inte-test"] = ""; + PAIMON_ASSIGN_OR_RAISE(auto catalog, Catalog::Create(root_path, new_options)); + PAIMON_RETURN_NOT_OK(catalog->CreateDatabase("foo", new_options, ignore_if_exists)); + ::ArrowSchema c_schema; + ScopeGuard guard([schema = &c_schema]() { ArrowSchemaRelease(schema); }); + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*schema, &c_schema)); + PAIMON_RETURN_NOT_OK(catalog->CreateTable(Identifier("foo", "bar"), &c_schema, + partition_keys, primary_keys, new_options, + ignore_if_exists)); + std::string table_path = PathUtil::JoinPath(root_path, "foo.db/bar"); + return Create(table_path, new_options, is_streaming_mode); + } + + static Result<std::unique_ptr<TestHelper>> Create( + const std::string& table_path, const std::map<std::string, std::string>& options, + bool is_streaming_mode) { + std::string file_system_identifier = "local"; + auto fs_iter = options.find(Options::FILE_SYSTEM); + if (fs_iter != options.end()) { + file_system_identifier = StringUtils::ToLowerCase(fs_iter->second); + } + PAIMON_ASSIGN_OR_RAISE(auto file_system, + FileSystemFactory::Get(file_system_identifier, table_path, options)); + std::string commit_user = "commit_user"; + WriteContextBuilder context_builder(table_path, commit_user); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<WriteContext> write_context, + context_builder.SetOptions(options) + .WithStreamingMode(is_streaming_mode) + .WithWriteId(12345) + .Finish()); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileStoreWrite> write, + FileStoreWrite::Create(std::move(write_context))); + std::map<std::string, std::string> new_options = options; + // only for test && only check the key + new_options["enable-pk-commit-in-inte-test"] = ""; + new_options["enable-object-store-commit-in-inte-test"] = ""; + CommitContextBuilder commit_context_builder(table_path, commit_user); + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr<CommitContext> commit_context, + commit_context_builder.SetOptions(new_options).IgnoreEmptyCommit(false).Finish()); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileStoreCommit> commit, + FileStoreCommit::Create(std::move(commit_context))); + return std::unique_ptr<TestHelper>(new TestHelper(std::move(file_system), std::move(write), + std::move(commit), commit_user, + table_path, options)); + } + + static Result<std::unique_ptr<RecordBatch>> MakeRecordBatch( + const std::shared_ptr<arrow::DataType>& data_type, const std::string& data_str, + const std::map<std::string, std::string>& partition_map, int32_t bucket, + const std::vector<RecordBatch::RowKind>& row_kinds) { + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + auto array, arrow::ipc::internal::json::ArrayFromJSON(data_type, data_str)); + ::ArrowArray arrow_array; + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*array, &arrow_array)); + RecordBatchBuilder batch_builder(&arrow_array); + return batch_builder.SetPartition(partition_map) + .SetBucket(bucket) + .SetRowKinds(row_kinds) + .Finish(); + } + + Result<std::vector<std::shared_ptr<CommitMessage>>> WriteAndCommit( + std::unique_ptr<RecordBatch>&& record_batch, int64_t commit_identifier, + const std::optional<std::vector<std::shared_ptr<CommitMessage>>>& + expected_commit_messages) { + std::vector<std::unique_ptr<RecordBatch>> batches; + batches.emplace_back(std::move(record_batch)); + return WriteAndCommit(std::move(batches), commit_identifier, expected_commit_messages); + } + + Result<std::vector<std::shared_ptr<CommitMessage>>> WriteAndCommit( + std::vector<std::unique_ptr<RecordBatch>>&& record_batches, int64_t commit_identifier, + const std::optional<std::vector<std::shared_ptr<CommitMessage>>>& + expected_commit_messages) { + for (auto& record_batch : record_batches) { + PAIMON_RETURN_NOT_OK(write_->Write(std::move(record_batch))); + } + PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<CommitMessage>> commit_messages, + write_->PrepareCommit(/*wait_compaction=*/false, commit_identifier)); + if (expected_commit_messages) { + CheckCommitMessages(expected_commit_messages.value(), commit_messages); + CheckExternalPath(commit_messages); + } + PAIMON_RETURN_NOT_OK(commit_->Commit(commit_messages, commit_identifier)); + return commit_messages; + } + + Result<std::vector<std::shared_ptr<Split>>> NewScan(StartupMode startup_mode, + std::optional<int64_t> snapshot_id, + bool is_streaming = true) { + ScanContextBuilder scan_context_builder(table_path_); + scan_context_builder.WithStreamingMode(is_streaming) + .SetOptions(options_) + .AddOption(Options::SCAN_MODE, startup_mode.ToString()); + if (snapshot_id) { + scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, + std::to_string(snapshot_id.value())); + } + PAIMON_ASSIGN_OR_RAISE(auto scan_context, scan_context_builder.Finish()); + PAIMON_ASSIGN_OR_RAISE(scan_, TableScan::Create(std::move(scan_context))); + return Scan(); + } + + Result<std::vector<std::shared_ptr<Split>>> Scan() { + if (scan_ == nullptr) { + return Status::Invalid("need call NewScan first"); + } + PAIMON_ASSIGN_OR_RAISE(auto result_plan, scan_->CreatePlan()); + return result_plan->Splits(); + } + + static Result<bool> CheckBlobsEqual(const std::vector<std::shared_ptr<Blob>>& result_blobs, + const std::vector<std::shared_ptr<Blob>>& expected_blobs, + const std::shared_ptr<FileSystem>& fs) { + if (result_blobs.size() != expected_blobs.size()) { + std::cout << "[result_blobs.size]: " << result_blobs.size() << std::endl; + std::cout << "[expected_blobs.size]: " << expected_blobs.size() << std::endl; + return false; + } + for (uint32_t i = 0; i < result_blobs.size(); ++i) { + PAIMON_ASSIGN_OR_RAISE(auto result_stream, result_blobs[i]->NewInputStream(fs)); + PAIMON_ASSIGN_OR_RAISE(auto expected_stream, expected_blobs[i]->NewInputStream(fs)); + + PAIMON_ASSIGN_OR_RAISE(uint64_t result_length, result_stream->Length()); + PAIMON_ASSIGN_OR_RAISE(uint64_t expected_length, expected_stream->Length()); + if (result_length != expected_length) { + auto result_descriptor_bytes = result_blobs[i]->ToDescriptor(GetDefaultPool()); + auto expected_descriptor_bytes = expected_blobs[i]->ToDescriptor(GetDefaultPool()); + PAIMON_ASSIGN_OR_RAISE( + auto result_descriptor, + BlobDescriptor::Deserialize(result_descriptor_bytes->data(), + result_descriptor_bytes->size())); + PAIMON_ASSIGN_OR_RAISE( + auto expected_descriptor, + BlobDescriptor::Deserialize(expected_descriptor_bytes->data(), + expected_descriptor_bytes->size())); + std::cout << "blobs[" << i << "]: " << std::endl; + std::cout << "[result_length(" << result_length << ") != expected_length(" + << expected_length << ")]" << std::endl; + std::cout << "[result_descriptor]: " << result_descriptor->ToString() << std::endl; + std::cout << "[expected_descriptor]: " << expected_descriptor->ToString() + << std::endl; + return false; + } + + std::vector<char> result_bytes(result_length, 0); + std::vector<char> expected_bytes(expected_length, 0); + PAIMON_RETURN_NOT_OK(result_stream->Read(result_bytes.data(), result_length)); + PAIMON_RETURN_NOT_OK(expected_stream->Read(expected_bytes.data(), expected_length)); + if (result_bytes != expected_bytes) { + std::cout << "blobs[" << i << "]: " << std::endl; + std::cout << "[result_bytes != expected_bytes]" << std::endl; + return false; + } + } + return true; + } + + static Result<std::vector<std::shared_ptr<Blob>>> ToBlobs( + const std::shared_ptr<arrow::StructArray>& blob_struct_array) { + std::vector<std::shared_ptr<Blob>> result_blobs; + auto child_array = blob_struct_array->field(0); + assert(blob_struct_array->num_fields() == 1); + assert(blob_struct_array->null_count() == 0); + assert(child_array->null_count() == 0); + assert(child_array->type_id() == arrow::Type::type::LARGE_BINARY); + + const auto& blob_array = + arrow::internal::checked_cast<const arrow::LargeBinaryArray&>(*child_array); + for (int64_t i = 0; i < blob_array.length(); ++i) { + std::string_view descriptor = blob_array.GetView(i); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Blob> blob, + Blob::FromDescriptor(descriptor.data(), descriptor.size())); + result_blobs.emplace_back(blob); + } + return result_blobs; + } + + // need to reconstruct the blob array, because the array in read result do not have blob meta + Result<std::shared_ptr<arrow::Array>> ReconstructBlobArray( + const std::shared_ptr<arrow::Array>& array, const std::shared_ptr<arrow::Schema>& schema) { + ::ArrowArray c_array; + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*array, &c_array)); + ::ArrowSchema new_c_schema; + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*schema, &new_c_schema)); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(auto new_array, + arrow::ImportArray(&c_array, &new_c_schema)); + return new_array; + } + + Result<bool> ReadAndCheckResultForBlobTable( + const std::shared_ptr<arrow::Schema>& all_columns_schema, + const std::vector<std::shared_ptr<Split>>& splits, const std::string& main_expected_json, + const std::vector<PAIMON_UNIQUE_PTR<Bytes>>& expected_blob_descriptors) { + ReadContextBuilder read_context_builder(table_path_); + 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(batch_reader.get())); + + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(auto concat_array, + arrow::Concatenate(read_result->chunks())); + PAIMON_ASSIGN_OR_RAISE(auto reconstruct_array, + ReconstructBlobArray(concat_array, all_columns_schema)); + PAIMON_ASSIGN_OR_RAISE( + auto separated_array, + BlobUtils::SeparateBlobArray( + std::dynamic_pointer_cast<arrow::StructArray>(reconstruct_array))); + + arrow::EqualOptions equal_options = arrow::EqualOptions::Defaults(); + + // check main columns + auto separated_schema = BlobUtils::SeparateBlobSchema(all_columns_schema); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + auto main_expected_array, + arrow::ipc::internal::json::ArrayFromJSON( + arrow::struct_(separated_schema.main_schema->fields()), main_expected_json)); + auto main_expected_chunk_array = std::make_shared<arrow::ChunkedArray>(main_expected_array); + bool main_equal = main_expected_chunk_array->Equals( + arrow::ChunkedArray(separated_array.main_array), equal_options.diff_sink(&std::cout)); + if (!main_equal) { + std::cout << "[expected_data_type]" << main_expected_chunk_array->type()->ToString() + << std::endl; + std::cout << "[actual_data_type]" << separated_array.main_array->type()->ToString() + << std::endl; + std::cout << "[expected]:" << main_expected_chunk_array->ToString() << std::endl; + std::cout << "[actual]: " << separated_array.main_array->ToString() << std::endl; + } + + // check blob column + std::vector<std::shared_ptr<Blob>> expected_blobs; + for (const auto& descriptor : expected_blob_descriptors) { + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Blob> blob, + Blob::FromDescriptor(descriptor->data(), descriptor->size())); + expected_blobs.emplace_back(blob); + } + PAIMON_ASSIGN_OR_RAISE(auto result_blobs, ToBlobs(separated_array.blob_array)); + PAIMON_ASSIGN_OR_RAISE(bool blob_equal, CheckBlobsEqual(result_blobs, expected_blobs, fs_)); + + table_read.reset(); + return main_equal && blob_equal; + } + + Result<bool> ReadAndCheckResult(const std::shared_ptr<arrow::DataType>& data_type, + const std::vector<std::shared_ptr<Split>>& splits, + const std::string& expected_result) { + ReadContextBuilder read_context_builder(table_path_); + 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(batch_reader.get())); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + auto expected_array, + arrow::ipc::internal::json::ArrayFromJSON(data_type, expected_result)); + auto expected_chunk_array = std::make_shared<arrow::ChunkedArray>(expected_array); + + arrow::EqualOptions equal_options = arrow::EqualOptions::Defaults(); + bool is_equal = + expected_chunk_array->Equals(read_result, equal_options.diff_sink(&std::cout)); + + if (!is_equal) { + std::cout << "[expected_data_type]" << expected_chunk_array->type()->ToString() + << std::endl; + std::cout << "[actual_data_type]" << read_result->type()->ToString() << std::endl; + std::cout << "[expected]:" << expected_chunk_array->ToString() << std::endl; + std::cout << "[actual]: " << read_result->ToString() << std::endl; + } + table_read.reset(); + return is_equal; + } + + Result<std::optional<Snapshot>> LatestSnapshot() const { + auto commit_impl = dynamic_cast<FileStoreCommitImpl*>(commit_.get()); + return commit_impl->snapshot_manager_->LatestSnapshotOfUser(commit_user_); + } Review Comment: `LatestSnapshot()` downcasts to `FileStoreCommitImpl` and dereferences internal members (`snapshot_manager_`) which are not part of the `FileStoreCommit` public API. This is fragile and can fail to compile (private/protected access) or crash if the implementation differs. ########## src/paimon/testing/utils/test_helper.h: ########## @@ -0,0 +1,443 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include <map> +#include <memory> +#include <string> +#include <utility> +#include <vector> + +#include "arrow/c/bridge.h" +#include "arrow/ipc/api.h" +#include "paimon/api.h" +#include "paimon/catalog/catalog.h" +#include "paimon/commit_context.h" +#include "paimon/common/data/blob_descriptor.h" +#include "paimon/common/data/blob_utils.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/common/utils/path_util.h" +#include "paimon/core/operation/append_only_file_store_write.h" +#include "paimon/core/operation/file_store_commit_impl.h" +#include "paimon/core/schema/schema_manager.h" +#include "paimon/core/table/sink/commit_message_impl.h" +#include "paimon/core/utils/file_store_path_factory.h" +#include "paimon/core/utils/snapshot_manager.h" +#include "paimon/data/blob.h" +#include "paimon/file_store_commit.h" +#include "paimon/file_store_write.h" +#include "paimon/fs/file_system_factory.h" +#include "paimon/record_batch.h" +#include "paimon/result.h" +#include "paimon/table/source/startup_mode.h" +#include "paimon/table/source/table_scan.h" +#include "paimon/testing/utils/read_result_collector.h" +#include "paimon/testing/utils/testharness.h" +#include "paimon/write_context.h" +namespace paimon::test { + +class TestHelper { + public: + static Result<std::unique_ptr<TestHelper>> Create( + const std::string& root_path, const std::shared_ptr<arrow::Schema>& schema, + const std::vector<std::string>& partition_keys, + const std::vector<std::string>& primary_keys, + const std::map<std::string, std::string>& options, bool is_streaming_mode, + bool ignore_if_exists = false) { + // only for test && only check the key + auto new_options = options; + new_options["enable-object-store-catalog-in-inte-test"] = ""; + PAIMON_ASSIGN_OR_RAISE(auto catalog, Catalog::Create(root_path, new_options)); + PAIMON_RETURN_NOT_OK(catalog->CreateDatabase("foo", new_options, ignore_if_exists)); + ::ArrowSchema c_schema; + ScopeGuard guard([schema = &c_schema]() { ArrowSchemaRelease(schema); }); + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*schema, &c_schema)); + PAIMON_RETURN_NOT_OK(catalog->CreateTable(Identifier("foo", "bar"), &c_schema, + partition_keys, primary_keys, new_options, + ignore_if_exists)); + std::string table_path = PathUtil::JoinPath(root_path, "foo.db/bar"); + return Create(table_path, new_options, is_streaming_mode); + } + + static Result<std::unique_ptr<TestHelper>> Create( + const std::string& table_path, const std::map<std::string, std::string>& options, + bool is_streaming_mode) { + std::string file_system_identifier = "local"; + auto fs_iter = options.find(Options::FILE_SYSTEM); + if (fs_iter != options.end()) { + file_system_identifier = StringUtils::ToLowerCase(fs_iter->second); + } + PAIMON_ASSIGN_OR_RAISE(auto file_system, + FileSystemFactory::Get(file_system_identifier, table_path, options)); + std::string commit_user = "commit_user"; + WriteContextBuilder context_builder(table_path, commit_user); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<WriteContext> write_context, + context_builder.SetOptions(options) + .WithStreamingMode(is_streaming_mode) + .WithWriteId(12345) + .Finish()); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileStoreWrite> write, + FileStoreWrite::Create(std::move(write_context))); + std::map<std::string, std::string> new_options = options; + // only for test && only check the key + new_options["enable-pk-commit-in-inte-test"] = ""; + new_options["enable-object-store-commit-in-inte-test"] = ""; + CommitContextBuilder commit_context_builder(table_path, commit_user); + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr<CommitContext> commit_context, + commit_context_builder.SetOptions(new_options).IgnoreEmptyCommit(false).Finish()); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileStoreCommit> commit, + FileStoreCommit::Create(std::move(commit_context))); + return std::unique_ptr<TestHelper>(new TestHelper(std::move(file_system), std::move(write), + std::move(commit), commit_user, + table_path, options)); + } + + static Result<std::unique_ptr<RecordBatch>> MakeRecordBatch( + const std::shared_ptr<arrow::DataType>& data_type, const std::string& data_str, + const std::map<std::string, std::string>& partition_map, int32_t bucket, + const std::vector<RecordBatch::RowKind>& row_kinds) { + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + auto array, arrow::ipc::internal::json::ArrayFromJSON(data_type, data_str)); + ::ArrowArray arrow_array; + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*array, &arrow_array)); + RecordBatchBuilder batch_builder(&arrow_array); + return batch_builder.SetPartition(partition_map) + .SetBucket(bucket) + .SetRowKinds(row_kinds) + .Finish(); + } + + Result<std::vector<std::shared_ptr<CommitMessage>>> WriteAndCommit( + std::unique_ptr<RecordBatch>&& record_batch, int64_t commit_identifier, + const std::optional<std::vector<std::shared_ptr<CommitMessage>>>& + expected_commit_messages) { + std::vector<std::unique_ptr<RecordBatch>> batches; + batches.emplace_back(std::move(record_batch)); + return WriteAndCommit(std::move(batches), commit_identifier, expected_commit_messages); + } + + Result<std::vector<std::shared_ptr<CommitMessage>>> WriteAndCommit( + std::vector<std::unique_ptr<RecordBatch>>&& record_batches, int64_t commit_identifier, + const std::optional<std::vector<std::shared_ptr<CommitMessage>>>& + expected_commit_messages) { + for (auto& record_batch : record_batches) { + PAIMON_RETURN_NOT_OK(write_->Write(std::move(record_batch))); + } + PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<CommitMessage>> commit_messages, + write_->PrepareCommit(/*wait_compaction=*/false, commit_identifier)); + if (expected_commit_messages) { + CheckCommitMessages(expected_commit_messages.value(), commit_messages); + CheckExternalPath(commit_messages); + } + PAIMON_RETURN_NOT_OK(commit_->Commit(commit_messages, commit_identifier)); + return commit_messages; + } + + Result<std::vector<std::shared_ptr<Split>>> NewScan(StartupMode startup_mode, + std::optional<int64_t> snapshot_id, + bool is_streaming = true) { + ScanContextBuilder scan_context_builder(table_path_); + scan_context_builder.WithStreamingMode(is_streaming) + .SetOptions(options_) + .AddOption(Options::SCAN_MODE, startup_mode.ToString()); + if (snapshot_id) { + scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, + std::to_string(snapshot_id.value())); + } + PAIMON_ASSIGN_OR_RAISE(auto scan_context, scan_context_builder.Finish()); + PAIMON_ASSIGN_OR_RAISE(scan_, TableScan::Create(std::move(scan_context))); + return Scan(); + } + + Result<std::vector<std::shared_ptr<Split>>> Scan() { + if (scan_ == nullptr) { + return Status::Invalid("need call NewScan first"); + } + PAIMON_ASSIGN_OR_RAISE(auto result_plan, scan_->CreatePlan()); + return result_plan->Splits(); + } + + static Result<bool> CheckBlobsEqual(const std::vector<std::shared_ptr<Blob>>& result_blobs, + const std::vector<std::shared_ptr<Blob>>& expected_blobs, + const std::shared_ptr<FileSystem>& fs) { + if (result_blobs.size() != expected_blobs.size()) { + std::cout << "[result_blobs.size]: " << result_blobs.size() << std::endl; + std::cout << "[expected_blobs.size]: " << expected_blobs.size() << std::endl; + return false; + } + for (uint32_t i = 0; i < result_blobs.size(); ++i) { + PAIMON_ASSIGN_OR_RAISE(auto result_stream, result_blobs[i]->NewInputStream(fs)); + PAIMON_ASSIGN_OR_RAISE(auto expected_stream, expected_blobs[i]->NewInputStream(fs)); + + PAIMON_ASSIGN_OR_RAISE(uint64_t result_length, result_stream->Length()); + PAIMON_ASSIGN_OR_RAISE(uint64_t expected_length, expected_stream->Length()); + if (result_length != expected_length) { + auto result_descriptor_bytes = result_blobs[i]->ToDescriptor(GetDefaultPool()); + auto expected_descriptor_bytes = expected_blobs[i]->ToDescriptor(GetDefaultPool()); + PAIMON_ASSIGN_OR_RAISE( + auto result_descriptor, + BlobDescriptor::Deserialize(result_descriptor_bytes->data(), + result_descriptor_bytes->size())); + PAIMON_ASSIGN_OR_RAISE( + auto expected_descriptor, + BlobDescriptor::Deserialize(expected_descriptor_bytes->data(), + expected_descriptor_bytes->size())); + std::cout << "blobs[" << i << "]: " << std::endl; + std::cout << "[result_length(" << result_length << ") != expected_length(" + << expected_length << ")]" << std::endl; + std::cout << "[result_descriptor]: " << result_descriptor->ToString() << std::endl; + std::cout << "[expected_descriptor]: " << expected_descriptor->ToString() + << std::endl; + return false; + } + + std::vector<char> result_bytes(result_length, 0); + std::vector<char> expected_bytes(expected_length, 0); + PAIMON_RETURN_NOT_OK(result_stream->Read(result_bytes.data(), result_length)); + PAIMON_RETURN_NOT_OK(expected_stream->Read(expected_bytes.data(), expected_length)); + if (result_bytes != expected_bytes) { + std::cout << "blobs[" << i << "]: " << std::endl; + std::cout << "[result_bytes != expected_bytes]" << std::endl; + return false; + } + } + return true; + } + + static Result<std::vector<std::shared_ptr<Blob>>> ToBlobs( + const std::shared_ptr<arrow::StructArray>& blob_struct_array) { + std::vector<std::shared_ptr<Blob>> result_blobs; + auto child_array = blob_struct_array->field(0); + assert(blob_struct_array->num_fields() == 1); + assert(blob_struct_array->null_count() == 0); + assert(child_array->null_count() == 0); + assert(child_array->type_id() == arrow::Type::type::LARGE_BINARY); + + const auto& blob_array = + arrow::internal::checked_cast<const arrow::LargeBinaryArray&>(*child_array); + for (int64_t i = 0; i < blob_array.length(); ++i) { + std::string_view descriptor = blob_array.GetView(i); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Blob> blob, + Blob::FromDescriptor(descriptor.data(), descriptor.size())); + result_blobs.emplace_back(blob); + } + return result_blobs; + } + + // need to reconstruct the blob array, because the array in read result do not have blob meta + Result<std::shared_ptr<arrow::Array>> ReconstructBlobArray( + const std::shared_ptr<arrow::Array>& array, const std::shared_ptr<arrow::Schema>& schema) { + ::ArrowArray c_array; + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*array, &c_array)); + ::ArrowSchema new_c_schema; + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*schema, &new_c_schema)); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(auto new_array, + arrow::ImportArray(&c_array, &new_c_schema)); + return new_array; + } + + Result<bool> ReadAndCheckResultForBlobTable( + const std::shared_ptr<arrow::Schema>& all_columns_schema, + const std::vector<std::shared_ptr<Split>>& splits, const std::string& main_expected_json, + const std::vector<PAIMON_UNIQUE_PTR<Bytes>>& expected_blob_descriptors) { + ReadContextBuilder read_context_builder(table_path_); + 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(batch_reader.get())); + + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(auto concat_array, + arrow::Concatenate(read_result->chunks())); + PAIMON_ASSIGN_OR_RAISE(auto reconstruct_array, + ReconstructBlobArray(concat_array, all_columns_schema)); + PAIMON_ASSIGN_OR_RAISE( + auto separated_array, + BlobUtils::SeparateBlobArray( + std::dynamic_pointer_cast<arrow::StructArray>(reconstruct_array))); + + arrow::EqualOptions equal_options = arrow::EqualOptions::Defaults(); + + // check main columns + auto separated_schema = BlobUtils::SeparateBlobSchema(all_columns_schema); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + auto main_expected_array, + arrow::ipc::internal::json::ArrayFromJSON( + arrow::struct_(separated_schema.main_schema->fields()), main_expected_json)); + auto main_expected_chunk_array = std::make_shared<arrow::ChunkedArray>(main_expected_array); + bool main_equal = main_expected_chunk_array->Equals( + arrow::ChunkedArray(separated_array.main_array), equal_options.diff_sink(&std::cout)); + if (!main_equal) { + std::cout << "[expected_data_type]" << main_expected_chunk_array->type()->ToString() + << std::endl; + std::cout << "[actual_data_type]" << separated_array.main_array->type()->ToString() + << std::endl; + std::cout << "[expected]:" << main_expected_chunk_array->ToString() << std::endl; + std::cout << "[actual]: " << separated_array.main_array->ToString() << std::endl; + } + + // check blob column + std::vector<std::shared_ptr<Blob>> expected_blobs; + for (const auto& descriptor : expected_blob_descriptors) { + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Blob> blob, + Blob::FromDescriptor(descriptor->data(), descriptor->size())); + expected_blobs.emplace_back(blob); + } + PAIMON_ASSIGN_OR_RAISE(auto result_blobs, ToBlobs(separated_array.blob_array)); + PAIMON_ASSIGN_OR_RAISE(bool blob_equal, CheckBlobsEqual(result_blobs, expected_blobs, fs_)); + + table_read.reset(); + return main_equal && blob_equal; + } + + Result<bool> ReadAndCheckResult(const std::shared_ptr<arrow::DataType>& data_type, + const std::vector<std::shared_ptr<Split>>& splits, + const std::string& expected_result) { + ReadContextBuilder read_context_builder(table_path_); + 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(batch_reader.get())); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + auto expected_array, + arrow::ipc::internal::json::ArrayFromJSON(data_type, expected_result)); + auto expected_chunk_array = std::make_shared<arrow::ChunkedArray>(expected_array); + + arrow::EqualOptions equal_options = arrow::EqualOptions::Defaults(); + bool is_equal = + expected_chunk_array->Equals(read_result, equal_options.diff_sink(&std::cout)); + + if (!is_equal) { + std::cout << "[expected_data_type]" << expected_chunk_array->type()->ToString() + << std::endl; + std::cout << "[actual_data_type]" << read_result->type()->ToString() << std::endl; + std::cout << "[expected]:" << expected_chunk_array->ToString() << std::endl; + std::cout << "[actual]: " << read_result->ToString() << std::endl; + } + table_read.reset(); + return is_equal; + } + + Result<std::optional<Snapshot>> LatestSnapshot() const { + auto commit_impl = dynamic_cast<FileStoreCommitImpl*>(commit_.get()); + return commit_impl->snapshot_manager_->LatestSnapshotOfUser(commit_user_); + } + + Result<std::optional<std::shared_ptr<TableSchema>>> LatestSchema() const { + auto commit_impl = dynamic_cast<FileStoreCommitImpl*>(commit_.get()); + return commit_impl->schema_manager_->Latest(); + } + + Result<std::string> PartitionStr(const BinaryRow& partition) const { + auto abstract_write = dynamic_cast<AbstractFileStoreWrite*>(write_.get()); + return abstract_write->file_store_path_factory_->GetPartitionString(partition); + } Review Comment: `PartitionStr()` downcasts to `AbstractFileStoreWrite` and accesses `file_store_path_factory_` directly. That member is `protected` in `AbstractFileStoreWrite`, so this is very likely to fail compilation. Consider generating the partition string via public utilities instead. ########## src/paimon/testing/utils/data_generator.cpp: ########## @@ -0,0 +1,314 @@ +/* + * 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 "paimon/testing/utils/data_generator.h" + +#include <algorithm> +#include <cassert> +#include <cmath> +#include <cstdlib> +#include <map> +#include <utility> + +#include "arrow/api.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 "fmt/format.h" +#include "paimon/common/data/binary_row_writer.h" +#include "paimon/common/data/binary_string.h" +#include "paimon/common/types/row_kind.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/core/schema/table_schema.h" +#include "paimon/core/utils/file_store_path_factory.h" + +namespace arrow { +class Array; +class ArrayBuilder; +} // namespace arrow +namespace paimon { +class MemoryPool; +} // namespace paimon + +namespace paimon::test { + +DataGenerator::DataGenerator(const std::shared_ptr<TableSchema>& table_schema, + const std::shared_ptr<MemoryPool>& memory_pool) + : table_schema_(table_schema), memory_pool_(memory_pool) { + assert(table_schema->Id() == 0); +} + +Status DataGenerator::WriteBinaryRow(const BinaryRow& src_row, int32_t src_field_id, + const std::shared_ptr<arrow::DataType>& src_type, + int32_t target_field_id, BinaryRowWriter* target_row_writer) { + arrow::Type::type type_id = src_type->id(); + switch (type_id) { + case arrow::Type::type::BOOL: { + target_row_writer->WriteBoolean(target_field_id, src_row.GetBoolean(src_field_id)); + break; + } + case arrow::Type::type::INT8: { + target_row_writer->WriteByte(target_field_id, src_row.GetByte(src_field_id)); + break; + } + case arrow::Type::type::INT16: { + target_row_writer->WriteShort(target_field_id, src_row.GetShort(src_field_id)); + break; + } + case arrow::Type::type::INT32: { + target_row_writer->WriteInt(target_field_id, src_row.GetInt(src_field_id)); + break; + } + case arrow::Type::type::INT64: { + target_row_writer->WriteLong(target_field_id, src_row.GetLong(src_field_id)); + break; + } + case arrow::Type::type::FLOAT: { + target_row_writer->WriteFloat(target_field_id, src_row.GetFloat(src_field_id)); + break; + } + case arrow::Type::type::DOUBLE: { + target_row_writer->WriteDouble(target_field_id, src_row.GetDouble(src_field_id)); + break; + } + case arrow::Type::type::STRING: { + target_row_writer->WriteString(target_field_id, src_row.GetString(src_field_id)); + break; + } + default: + return Status::Invalid( + fmt::format("type {} not support in write partial row", src_type->ToString())); + } + return Status::OK(); +} + +Result<BinaryRow> DataGenerator::ExtractPartialRow(const BinaryRow& binary_row, + const std::vector<DataField>& partition_fields) { + BinaryRow partial_row(static_cast<int32_t>(partition_fields.size())); + BinaryRowWriter writer(&partial_row, /*initial_size=*/0, memory_pool_.get()); + for (size_t field_idx = 0; field_idx < partition_fields.size(); field_idx++) { + int32_t id = partition_fields[field_idx].Id(); + auto type = partition_fields[field_idx].Type(); + PAIMON_RETURN_NOT_OK(WriteBinaryRow(binary_row, id, type, field_idx, &writer)); + } Review Comment: `ExtractPartialRow()` uses `DataField::Id()` as the source index into `BinaryRow` (`GetInt`, `GetString`, etc.). `BinaryRow` accessors are positional, while `DataField::Id()` is a schema field-id that can diverge from the column position when the schema contains nested fields (IDs are assigned recursively). This can produce wrong partition/bucket rows for nested schemas. ########## src/paimon/testing/utils/dict_array_converter.h: ########## @@ -0,0 +1,140 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include <memory> + +#include "arrow/api.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/result.h" + +namespace paimon::test { +class DictArrayConverter { + public: + DictArrayConverter() = delete; + ~DictArrayConverter() = delete; + + // deep copy dictionary array to string array/binary array + static Result<std::shared_ptr<arrow::Array>> ConvertDictArray( + const std::shared_ptr<arrow::Array>& array, arrow::MemoryPool* pool) { + arrow::Type::type kind = array->type_id(); + switch (kind) { + case arrow::Type::type::STRUCT: { + // convert array + auto struct_array = + arrow::internal::checked_pointer_cast<arrow::StructArray>(array); + arrow::ArrayVector new_children; + std::size_t size = struct_array->fields().size(); + for (size_t i = 0; i < size; i++) { + std::shared_ptr<arrow::Array> child = struct_array->field(static_cast<int>(i)); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> new_child, + ConvertDictArray(child, pool)); + new_children.push_back(new_child); + } + + // convert type + arrow::FieldVector fields; + fields.reserve(new_children.size()); + for (size_t i = 0; i < new_children.size(); i++) { + // Note: For test consistency, intentionally left nullable unspecified, as ORC + // discard nullable information, making it impossible to align. + // Moreover, this detail is currently not important for users. + fields.push_back(arrow::field(struct_array->type()->field(i)->name(), + new_children[i]->type())); + } + + return std::make_shared<arrow::StructArray>(arrow::struct_(fields), + struct_array->length(), new_children, + struct_array->null_bitmap()); + } + case arrow::Type::type::LIST: { + auto list_array = arrow::internal::checked_pointer_cast<arrow::ListArray>(array); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> value_array, + ConvertDictArray(list_array->values(), pool)); + return std::make_shared<arrow::ListArray>( + arrow::list(value_array->type()), list_array->length(), + list_array->value_offsets(), value_array, list_array->null_bitmap(), + list_array->null_count(), list_array->offset()); + } + case arrow::Type::type::MAP: { + auto map_array = arrow::internal::checked_pointer_cast<arrow::MapArray>(array); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> key_array, + ConvertDictArray(map_array->keys(), pool)); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> item_array, + ConvertDictArray(map_array->items(), pool)); + auto map_type = + arrow::internal::checked_pointer_cast<arrow::MapType>(map_array->type()); + auto new_map_type = std::make_shared<arrow::MapType>( + key_array->type(), item_array->type(), map_type->keys_sorted()); + return std::make_shared<arrow::MapArray>( + new_map_type, map_array->length(), map_array->value_offsets(), key_array, + item_array, map_array->null_bitmap(), map_array->null_count(), + map_array->offset()); + } + case arrow::Type::type::DICTIONARY: { + auto dict_array = + arrow::internal::checked_pointer_cast<arrow::DictionaryArray>(array); + auto dict_type = arrow::internal::checked_pointer_cast<arrow::DictionaryType>( + dict_array->type()); + auto value_type_id = dict_type->value_type()->id(); + auto index_type_id = dict_type->index_type()->id(); + if (value_type_id == arrow::Type::type::STRING && + index_type_id == arrow::Type::type::INT32) { + return ConvertDictionaryArrayToBinaryArray< + arrow::StringArray, arrow::Int32Array, arrow::StringBuilder>(dict_array, + pool); + } else if (value_type_id == arrow::Type::type::LARGE_STRING && + index_type_id == arrow::Type::type::INT64) { + return ConvertDictionaryArrayToBinaryArray< + arrow::LargeStringArray, arrow::Int64Array, arrow::StringBuilder>( + dict_array, pool); Review Comment: For `[LARGE_STRING, INT64]` dictionary arrays, the conversion uses `arrow::StringBuilder`, which produces a `string` array (32-bit offsets) rather than a `large_string` array (64-bit offsets). This can lead to type mismatches or overflow for large offsets. Use `arrow::LargeStringBuilder` here to preserve the large-string type. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
