| /* | |
| * 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 <cstddef> | |
| #include <cstdint> | |
| #include <iostream> | |
| #include <map> | |
| #include <memory> | |
| #include <optional> | |
| #include <set> | |
| #include <string> | |
| #include <tuple> | |
| #include <utility> | |
| #include <vector> | |
| #include "arrow/api.h" | |
| #include "arrow/c/bridge.h" | |
| #include "arrow/ipc/json_simple.h" | |
| #include "fmt/format.h" | |
| #include "fmt/ranges.h" | |
| #include "gtest/gtest.h" | |
| #include "paimon/bucket/bucket_id_calculator.h" | |
| #include "paimon/common/factories/io_hook.h" | |
| #include "paimon/common/io/cache/lru_cache.h" | |
| #include "paimon/common/table/special_fields.h" | |
| #include "paimon/common/types/data_field.h" | |
| #include "paimon/common/utils/checked_cast.h" | |
| #include "paimon/common/utils/date_time_utils.h" | |
| #include "paimon/common/utils/path_util.h" | |
| #include "paimon/common/utils/scope_guard.h" | |
| #include "paimon/common/utils/string_utils.h" | |
| #include "paimon/core/io/data_file_meta.h" | |
| #include "paimon/core/table/source/data_split_impl.h" | |
| #include "paimon/core/table/source/deletion_file.h" | |
| #include "paimon/core/table/source/key_value_table_read.h" | |
| #include "paimon/defs.h" | |
| #include "paimon/executor.h" | |
| #include "paimon/fs/file_system.h" | |
| #include "paimon/fs/local/local_file_system.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/plan.h" | |
| #include "paimon/table/source/scan_metrics.h" | |
| #include "paimon/table/source/startup_mode.h" | |
| #include "paimon/table/source/table_read.h" | |
| #include "paimon/table/source/table_scan.h" | |
| #include "paimon/testing/utils/counting_cache_test_utils.h" | |
| #include "paimon/testing/utils/io_exception_helper.h" | |
| #include "paimon/testing/utils/read_result_collector.h" | |
| #include "paimon/testing/utils/test_helper.h" | |
| #include "paimon/testing/utils/testharness.h" | |
| #include "paimon/testing/utils/timezone_guard.h" | |
| namespace paimon { | |
| class DataSplit; | |
| } // namespace paimon | |
| namespace paimon::test { | |
| using ScanAndReadParam = std::tuple<std::string, bool, bool>; | |
| class ScanAndReadInteTest : public testing::Test, | |
| public ::testing::WithParamInterface<ScanAndReadParam> { | |
| public: | |
| void CheckStreamScanResult( | |
| const std::unique_ptr<TableScan>& table_scan, const std::unique_ptr<TableRead>& table_read, | |
| const std::vector<std::optional<int64_t>>& expected_snapshot_ids, | |
| const std::vector<std::shared_ptr<arrow::ChunkedArray>>& expected_array) const { | |
| size_t scan_id = 0; | |
| while (true) { | |
| ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> result_plan, table_scan->CreatePlan()); | |
| if (scan_id == expected_snapshot_ids.size()) { | |
| // no snapshot | |
| ASSERT_EQ(std::nullopt, result_plan->SnapshotId()); | |
| ASSERT_TRUE(result_plan->Splits().empty()); | |
| break; | |
| } | |
| ASSERT_EQ(result_plan->SnapshotId(), expected_snapshot_ids[scan_id]); | |
| auto splits = result_plan->Splits(); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(splits)); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| if (expected_array[scan_id]) { | |
| ASSERT_TRUE(read_result); | |
| ASSERT_TRUE(expected_array[scan_id]->type()->Equals(read_result->type())); | |
| ASSERT_TRUE(expected_array[scan_id]->Equals(read_result)) | |
| << read_result->ToString() << std::endl | |
| << "expected" << expected_array[scan_id]->ToString(); | |
| } else { | |
| ASSERT_FALSE(read_result); | |
| } | |
| scan_id++; | |
| } | |
| } | |
| void PrintPlan(const std::shared_ptr<Plan>& plan) const { | |
| std::string snapshot_str = | |
| plan->SnapshotId() ? std::to_string(plan->SnapshotId().value()) : "null"; | |
| std::cout << "snapshot id=" << snapshot_str << std::endl; | |
| const auto& splits = plan->Splits(); | |
| for (const auto& split : splits) { | |
| auto split_impl = std::dynamic_pointer_cast<DataSplitImpl>(split); | |
| std::cout << split_impl->ToString() << std::endl; | |
| } | |
| } | |
| void CheckPostponeFile(const std::string& root_path, | |
| const std::vector<std::string>& subdirs) const { | |
| std::vector<BasicFileStatus> status_list; | |
| auto file_system = std::make_shared<LocalFileSystem>(); | |
| for (const auto& dir : subdirs) { | |
| ASSERT_OK(file_system->ListDir(PathUtil::JoinPath(root_path, dir), &status_list)); | |
| } | |
| ASSERT_FALSE(status_list.empty()); | |
| for (const auto& file_status : status_list) { | |
| std::string path = file_status.GetPath(); | |
| ASSERT_TRUE(path.find("-u-") != std::string::npos); | |
| ASSERT_TRUE(path.find("-s-") != std::string::npos); | |
| ASSERT_TRUE(path.find("-w-") != std::string::npos); | |
| // writer id | |
| ASSERT_TRUE(path.find("12345") != std::string::npos); | |
| } | |
| } | |
| void AddReadOptionsForPrefetch(ReadContextBuilder* read_context_builder) { | |
| read_context_builder->AddOption("test.enable-adaptive-prefetch-strategy", "false"); | |
| if (EnablePrefetch()) { | |
| read_context_builder->EnablePrefetch(true).SetPrefetchBatchCount(3); | |
| } | |
| } | |
| Result<std::unique_ptr<ScanContext>> FinishScanContext(ScanContextBuilder& builder) { | |
| if (EnableSnapshotLiveManifestCache()) { | |
| if (!snapshot_live_manifest_cache_) { | |
| snapshot_live_manifest_cache_ = | |
| std::make_shared<LruCache>(/*max_weight=*/64 * 1024 * 1024); | |
| } | |
| builder.AddOption(Options::SCAN_MANIFEST_ENTRY_CACHE_MAX_SNAPSHOTS, "3") | |
| .WithCache(snapshot_live_manifest_cache_); | |
| } | |
| return builder.Finish(); | |
| } | |
| std::string FileFormat() const { | |
| return std::get<0>(GetParam()); | |
| } | |
| bool EnablePrefetch() const { | |
| return std::get<1>(GetParam()); | |
| } | |
| bool EnableSnapshotLiveManifestCache() const { | |
| return std::get<2>(GetParam()); | |
| } | |
| void AdjustSplitWithExternalPath(const std::string& src_path, const std::string& target_path, | |
| bool adjust_index, | |
| std::vector<std::shared_ptr<Split>>* splits_ptr) { | |
| // adjust external path from src_path to target_path | |
| auto& splits = *splits_ptr; | |
| for (auto& split : splits) { | |
| auto data_split = std::dynamic_pointer_cast<DataSplitImpl>(split); | |
| ASSERT_TRUE(data_split); | |
| auto& data_files = data_split->data_files_; | |
| for (auto& file : data_files) { | |
| auto& external_path = file->external_path; | |
| if (external_path) { | |
| external_path = | |
| StringUtils::Replace(external_path.value(), src_path, target_path); | |
| } | |
| } | |
| if (adjust_index) { | |
| auto& deletion_files = data_split->data_deletion_files_; | |
| for (auto& file : deletion_files) { | |
| if (file) { | |
| file.value().path = | |
| StringUtils::Replace(file.value().path, src_path, target_path); | |
| } | |
| } | |
| } | |
| } | |
| } | |
| private: | |
| std::shared_ptr<Cache> snapshot_live_manifest_cache_; | |
| std::shared_ptr<arrow::StructType> arrow_data_type_ = | |
| std::dynamic_pointer_cast<arrow::StructType>(DataField::ConvertDataFieldsToArrowStructType( | |
| {SpecialFields::ValueKind(), DataField(0, arrow::field("f0", arrow::utf8())), | |
| DataField(1, arrow::field("f1", arrow::int32())), | |
| DataField(2, arrow::field("f2", arrow::int32())), | |
| DataField(3, arrow::field("f3", arrow::float64()))})); | |
| }; | |
| std::vector<ScanAndReadParam> GetTestValuesForScanAndReadInteTest() { | |
| std::vector<ScanAndReadParam> values; | |
| for (bool enable_snapshot_live_manifest_cache : {false, true}) { | |
| values.emplace_back("parquet", false, enable_snapshot_live_manifest_cache); | |
| values.emplace_back("parquet", true, enable_snapshot_live_manifest_cache); | |
| #ifdef PAIMON_ENABLE_ORC | |
| values.emplace_back("orc", false, enable_snapshot_live_manifest_cache); | |
| values.emplace_back("orc", true, enable_snapshot_live_manifest_cache); | |
| #endif | |
| } | |
| return values; | |
| } | |
| INSTANTIATE_TEST_SUITE_P(FileFormatAndEnablePaimonPrefetch, ScanAndReadInteTest, | |
| ::testing::ValuesIn(GetTestValuesForScanAndReadInteTest())); | |
| TEST_P(ScanAndReadInteTest, TestWithAppendSnapshotIOException) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = GetDataDir() + "/" + file_format + "/append_09.db/append_09"; | |
| bool run_complete = false; | |
| auto io_hook = IOHook::GetInstance(); | |
| for (size_t i = 0; i < 500; i++) { | |
| ScopeGuard guard([&io_hook]() { io_hook->Clear(); }); | |
| io_hook->Reset(i, IOHook::Mode::RETURN_ERROR); | |
| // scan | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "1"); | |
| Result<std::unique_ptr<ScanContext>> scan_context = FinishScanContext(scan_context_builder); | |
| CHECK_HOOK_STATUS(scan_context.status(), i); | |
| Result<std::unique_ptr<TableScan>> table_scan = | |
| TableScan::Create(std::move(scan_context).value()); | |
| CHECK_HOOK_STATUS(table_scan.status(), i); | |
| Result<std::shared_ptr<Plan>> result_plan = table_scan.value()->CreatePlan(); | |
| CHECK_HOOK_STATUS(result_plan.status(), i); | |
| ASSERT_EQ(result_plan.value()->SnapshotId().value(), 1); | |
| auto splits = result_plan.value()->Splits(); | |
| ASSERT_EQ(3, splits.size()); | |
| // read | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, | |
| read_context_builder.Finish()); | |
| Result<std::unique_ptr<TableRead>> table_read = TableRead::Create(std::move(read_context)); | |
| CHECK_HOOK_STATUS(table_read.status(), i); | |
| Result<std::unique_ptr<BatchReader>> batch_reader = | |
| table_read.value()->CreateReader(splits); | |
| CHECK_HOOK_STATUS(batch_reader.status(), i); | |
| std::unique_ptr<BatchReader> owned_reader = std::move(batch_reader).value(); | |
| auto read_result = ReadResultCollector::CollectResult(std::move(owned_reader)); | |
| CHECK_HOOK_STATUS(read_result.status(), i); | |
| // check result | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Alice", 10, 1, 11.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Tony", 10, 0, 14.1], | |
| [0, "Lucy", 20, 1, 14.1] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result.value())) << read_result.value()->ToString(); | |
| run_complete = true; | |
| break; | |
| } | |
| ASSERT_TRUE(run_complete); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithPkSnapshotIOException) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = GetDataDir() + "/" + file_format + | |
| "/pk_table_scan_and_read_dv.db/pk_table_scan_and_read_dv/"; | |
| bool run_complete = false; | |
| auto io_hook = IOHook::GetInstance(); | |
| for (size_t i = 0; i < 800; i++) { | |
| ScopeGuard guard([&io_hook]() { io_hook->Clear(); }); | |
| io_hook->Reset(i, IOHook::Mode::RETURN_ERROR); | |
| // scan | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "6"); | |
| Result<std::unique_ptr<ScanContext>> scan_context = FinishScanContext(scan_context_builder); | |
| CHECK_HOOK_STATUS(scan_context.status(), i); | |
| Result<std::unique_ptr<TableScan>> table_scan = | |
| TableScan::Create(std::move(scan_context).value()); | |
| CHECK_HOOK_STATUS(table_scan.status(), i); | |
| Result<std::shared_ptr<Plan>> result_plan = table_scan.value()->CreatePlan(); | |
| CHECK_HOOK_STATUS(result_plan.status(), i); | |
| ASSERT_EQ(result_plan.value()->SnapshotId().value(), 6); | |
| auto splits = result_plan.value()->Splits(); | |
| ASSERT_EQ(3, splits.size()); | |
| // read | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, | |
| read_context_builder.Finish()); | |
| Result<std::unique_ptr<TableRead>> table_read = TableRead::Create(std::move(read_context)); | |
| CHECK_HOOK_STATUS(table_read.status(), i); | |
| Result<std::unique_ptr<BatchReader>> batch_reader = | |
| table_read.value()->CreateReader(splits); | |
| CHECK_HOOK_STATUS(batch_reader.status(), i); | |
| std::unique_ptr<BatchReader> owned_reader = std::move(batch_reader).value(); | |
| auto read_result = ReadResultCollector::CollectResult(std::move(owned_reader)); | |
| CHECK_HOOK_STATUS(read_result.status(), i); | |
| // check result | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Two roads diverged in a wood, and I took the one less traveled by, And that has made all the difference.", 10, 1, 11.0], | |
| [0, "Alice", 10, 1, 19.1], | |
| [0, "Alex", 10, 0, 16.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "David", 10, 0, 17.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Lucy", 20, 1, 14.1], | |
| [0, "Paul", 20, 1, 18.1] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result.value())) << read_result.value()->ToString(); | |
| run_complete = true; | |
| break; | |
| } | |
| ASSERT_TRUE(run_complete); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithAppendSnapshot1) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = GetDataDir() + "/" + file_format + "/append_09.db/append_09"; | |
| // scan | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "1"); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 1); | |
| auto splits = result_plan->Splits(); | |
| ASSERT_EQ(3, splits.size()); | |
| // read | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(splits)); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Alice", 10, 1, 11.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Tony", 10, 0, 14.1], | |
| [0, "Lucy", 20, 1, 14.1] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| ASSERT_OK_AND_ASSIGN(auto count_reader, table_read->CreateCountReader(splits)); | |
| ASSERT_OK_AND_ASSIGN(int64_t count, count_reader->CountRows()); | |
| ASSERT_EQ(count, read_result->length()); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithAppendSnapshot3) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = GetDataDir() + "/" + file_format + "/append_09.db/append_09"; | |
| // scan | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "3"); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 3); | |
| auto splits = result_plan->Splits(); | |
| ASSERT_EQ(3, splits.size()); | |
| // read | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(splits)); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Alice", 10, 1, 11.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Tony", 10, 0, 14.1], | |
| [0, "Emily", 10, 0, 15.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "Alex", 10, 0, 16.1], | |
| [0, "David", 10, 0, 17.1], | |
| [0, "Lucy", 20, 1, 14.1], | |
| [0, "Paul", 20, 1, null] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| } | |
| TEST(SelectiveManifestDecodeInteTest, TestPartitionedPointLookup) { | |
| // Scale the many-partition/many-bucket point-lookup workload down for CI. | |
| constexpr int32_t kPartitions = 4; | |
| constexpr int32_t kBuckets = 32; | |
| constexpr int32_t kCandidateKeys = 1024; | |
| const std::string payload(16 * 1024, 'x'); | |
| auto dir = UniqueTestDirectory::Create("local"); | |
| ASSERT_TRUE(dir); | |
| arrow::FieldVector fields = {arrow::field("p", arrow::int32()), | |
| arrow::field("rowkey", arrow::utf8()), | |
| arrow::field("payload", arrow::utf8())}; | |
| auto schema = arrow::schema(fields); | |
| auto data_type = arrow::struct_(fields); | |
| std::map<std::string, std::string> options = {{Options::FILE_FORMAT, "orc"}, | |
| {Options::BUCKET, fmt::format("{}", kBuckets)}, | |
| {Options::BUCKET_KEY, "rowkey"}, | |
| {Options::MANIFEST_COMPRESSION, "zstd"}, | |
| {Options::MANIFEST_TARGET_FILE_SIZE, "64 mb"}}; | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<TestHelper> helper, | |
| TestHelper::Create(dir->Str(), schema, {"p"}, {}, options, false)); | |
| const std::string table_path = PathUtil::JoinPath(dir->Str(), "foo.db/bar"); | |
| std::vector<std::string> candidate_rows; | |
| for (int32_t i = 0; i < kCandidateKeys; ++i) { | |
| candidate_rows.push_back(fmt::format(R"(["key{:04}"])", i)); | |
| } | |
| auto key_array = | |
| arrow::ipc::internal::json::ArrayFromJSON( | |
| arrow::struct_({fields[1]}), fmt::format("[{}]", fmt::join(candidate_rows, ","))) | |
| .ValueOrDie(); | |
| ArrowArray c_keys; | |
| ArrowSchema c_key_schema; | |
| ASSERT_TRUE(arrow::ExportArray(*key_array, &c_keys, &c_key_schema).ok()); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<BucketIdCalculator> calculator, | |
| BucketIdCalculator::Create(false, kBuckets, GetDefaultPool())); | |
| std::vector<int32_t> bucket_ids(kCandidateKeys); | |
| ASSERT_OK(calculator->CalculateBucketIds(&c_keys, &c_key_schema, bucket_ids.data())); | |
| std::vector<std::string> keys(kBuckets); | |
| for (int32_t i = 0; i < kCandidateKeys; ++i) { | |
| if (keys[bucket_ids[i]].empty()) { | |
| keys[bucket_ids[i]] = fmt::format("key{:04}", i); | |
| } | |
| } | |
| std::vector<std::unique_ptr<RecordBatch>> batches; | |
| for (int32_t p = 0; p < kPartitions; ++p) { | |
| for (int32_t bucket = 0; bucket < kBuckets; ++bucket) { | |
| ASSERT_FALSE(keys[bucket].empty()); | |
| ASSERT_OK_AND_ASSIGN( | |
| std::unique_ptr<RecordBatch> batch, | |
| TestHelper::MakeRecordBatch( | |
| data_type, fmt::format(R"([[{},"{}","{}"]])", p, keys[bucket], payload), | |
| {{"p", fmt::format("{}", p)}}, bucket, {})); | |
| batches.push_back(std::move(batch)); | |
| } | |
| } | |
| ASSERT_OK(helper->WriteAndCommit(std::move(batches), 0, std::nullopt)); | |
| const std::string& lookup_key = keys.back(); | |
| auto predicate = | |
| PredicateBuilder::Equal(1, "rowkey", FieldType::STRING, | |
| Literal(FieldType::STRING, lookup_key.data(), lookup_key.size())); | |
| ASSERT_OK_AND_ASSIGN(std::shared_ptr<Executor> executor, CreateDefaultExecutor(1)); | |
| // Plans and cache values can outlive a scan and reference its allocator. | |
| // Keep every measured pool alive until all plans and caches have been destroyed. | |
| std::vector<std::shared_ptr<MemoryPool>> scan_pools; | |
| struct ScanResult { | |
| std::shared_ptr<Plan> plan; | |
| uint64_t cache_enabled; | |
| uint64_t cache_hit; | |
| int64_t peak_bytes; | |
| }; | |
| auto scan = [&](bool lazy_decode, const std::shared_ptr<Cache>& cache, | |
| bool entry_cache) -> Result<ScanResult> { | |
| std::shared_ptr<MemoryPool> pool = GetMemoryPool(); | |
| scan_pools.push_back(pool); | |
| ScanContextBuilder builder(table_path); | |
| builder.WithMemoryPool(pool) | |
| .WithExecutor(executor) | |
| .WithCache(cache) | |
| .AddOption(Options::SCAN_MANIFEST_ENTRY_CACHE_MAX_SNAPSHOTS, entry_cache ? "3" : "0") | |
| .AddOption(Options::SCAN_MANIFEST_ENTRY_LAZY_DECODE_ENABLED, | |
| lazy_decode ? "true" : "false") | |
| .AddOption(Options::READ_BATCH_SIZE, "128"); | |
| if (entry_cache) { | |
| // No explicit bucket or partition filter: the predicate must select the bucket. | |
| builder.SetPredicate(predicate); | |
| } | |
| PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ScanContext> context, builder.Finish()); | |
| PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<TableScan> table_scan, | |
| TableScan::Create(std::move(context))); | |
| PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Plan> plan, table_scan->CreatePlan()); | |
| PAIMON_ASSIGN_OR_RAISE(uint64_t enabled, table_scan->GetMetrics()->GetCounter( | |
| ScanMetrics::LAST_SNAPSHOT_CACHE_ENABLED)); | |
| PAIMON_ASSIGN_OR_RAISE(uint64_t hit, table_scan->GetMetrics()->GetCounter( | |
| ScanMetrics::LAST_SNAPSHOT_CACHE_HIT)); | |
| return ScanResult{plan, enabled, hit, pool->MaxMemoryUsage()}; | |
| }; | |
| auto check_read = [&](const ScanResult& result, int32_t partition_count, bool historical) { | |
| ASSERT_EQ(result.cache_enabled, 1); | |
| ASSERT_EQ(result.plan->Splits().size(), partition_count); | |
| for (const auto& split : result.plan->Splits()) { | |
| auto data_split = std::dynamic_pointer_cast<DataSplit>(split); | |
| ASSERT_TRUE(data_split); | |
| ASSERT_TRUE(data_split->Bucket() == kBuckets - 1 || | |
| (historical && data_split->Bucket() == kBuckets / 2 - 1)); | |
| ASSERT_EQ(data_split->GetFileList().size(), 1); | |
| } | |
| ReadContextBuilder builder(table_path); | |
| builder.SetPredicate(predicate).EnablePredicateFilter(true); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> context, builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableRead> read, | |
| TableRead::Create(std::move(context))); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<BatchReader> reader, | |
| read->CreateReader(result.plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> rows, | |
| ReadResultCollector::CollectResult(std::move(reader))); | |
| ASSERT_TRUE(rows); | |
| ASSERT_EQ(rows->length(), partition_count); | |
| std::set<int32_t> partitions; | |
| for (const auto& chunk : rows->chunks()) { | |
| auto values = std::dynamic_pointer_cast<arrow::StructArray>(chunk); | |
| ASSERT_TRUE(values); | |
| auto p = std::dynamic_pointer_cast<arrow::Int32Array>(values->GetFieldByName("p")); | |
| auto key = | |
| std::dynamic_pointer_cast<arrow::StringArray>(values->GetFieldByName("rowkey")); | |
| auto data = | |
| std::dynamic_pointer_cast<arrow::StringArray>(values->GetFieldByName("payload")); | |
| ASSERT_TRUE(p && key && data); | |
| for (int64_t i = 0; i < values->length(); ++i) { | |
| ASSERT_TRUE(partitions.insert(p->Value(i)).second); | |
| ASSERT_EQ(key->GetString(i), lookup_key); | |
| ASSERT_EQ(data->GetString(i), payload); | |
| } | |
| } | |
| for (int32_t p = 0; p < partition_count; ++p) { | |
| ASSERT_EQ(partitions.count(p), 1); | |
| } | |
| }; | |
| std::vector<std::shared_ptr<CountingRoutingCache>> caches; | |
| std::vector<int64_t> peaks; | |
| for (bool lazy_decode : {false, true}) { | |
| auto cache = std::make_shared<CountingRoutingCache>( | |
| std::map<CacheKind, int64_t>{{CacheKind::MANIFEST, 64 * 1024 * 1024}, | |
| {CacheKind::SNAPSHOT_LIVE_MANIFEST, 64 * 1024 * 1024}}); | |
| caches.push_back(cache); | |
| // Warm only raw manifest bytes, with a separate pool, before measuring either mode. | |
| ASSERT_OK_AND_ASSIGN(ScanResult warm, scan(false, cache, false)); | |
| ASSERT_EQ(warm.cache_enabled, 0); | |
| size_t file_count = 0; | |
| for (const auto& split : warm.plan->Splits()) { | |
| auto data_split = std::dynamic_pointer_cast<DataSplit>(split); | |
| ASSERT_TRUE(data_split); | |
| file_count += data_split->GetFileList().size(); | |
| } | |
| ASSERT_EQ(file_count, kPartitions * kBuckets); | |
| const int64_t supplier_calls = cache->SupplierCallCount(CacheKind::MANIFEST); | |
| ASSERT_GT(supplier_calls, 0); | |
| ASSERT_OK_AND_ASSIGN(ScanResult measured, scan(lazy_decode, cache, true)); | |
| ASSERT_EQ(measured.cache_hit, 0); | |
| ASSERT_NO_FATAL_FAILURE(check_read(measured, kPartitions, false)); | |
| ASSERT_EQ(cache->SupplierCallCount(CacheKind::MANIFEST), supplier_calls); | |
| peaks.push_back(measured.peak_bytes); | |
| RecordProperty(lazy_decode ? "selective_peak_bytes" : "baseline_peak_bytes", | |
| fmt::format("{}", measured.peak_bytes)); | |
| ASSERT_OK_AND_ASSIGN(ScanResult hit, scan(lazy_decode, cache, true)); | |
| ASSERT_EQ(hit.cache_hit, 1); | |
| ASSERT_NO_FATAL_FAILURE(check_read(hit, kPartitions, false)); | |
| // Manifest lists are still consulted, but their bytes must also come from the cache. | |
| ASSERT_EQ(cache->SupplierCallCount(CacheKind::MANIFEST), supplier_calls); | |
| } | |
| // Selective decoding also applies when only manifest entries, not raw bytes, are cached. | |
| auto entry_cache_only = | |
| std::make_shared<CountingRoutingCache>(CacheKind::SNAPSHOT_LIVE_MANIFEST, 64 * 1024 * 1024); | |
| ASSERT_OK_AND_ASSIGN(ScanResult uncached_manifest, scan(true, entry_cache_only, true)); | |
| ASSERT_EQ(uncached_manifest.cache_hit, 0); | |
| ASSERT_NO_FATAL_FAILURE(check_read(uncached_manifest, kPartitions, false)); | |
| ASSERT_GT(entry_cache_only->GetCount(CacheKind::MANIFEST), 0); | |
| ASSERT_EQ(entry_cache_only->SupplierCallCount(CacheKind::MANIFEST), 0); | |
| RecordProperty("uncached_manifest_peak_bytes", fmt::format("{}", uncached_manifest.peak_bytes)); | |
| // Wide, real column statistics make full Arrow materialization observable without timing gates. | |
| ASSERT_LT(peaks[1], peaks[0] / 2); | |
| ASSERT_LT(uncached_manifest.peak_bytes, peaks[0] / 2); | |
| // Change both schema ID and bucket count, then append the same key in a new partition. | |
| // Old files must still be read from bucket 31 while the new file belongs to bucket 15. | |
| helper.reset(); | |
| options[Options::BUCKET] = fmt::format("{}", kBuckets / 2); | |
| ASSERT_OK(TestHelper::WriteNextSchema( | |
| dir->GetFileSystem(), table_path, | |
| {DataField(0, fields[0]), DataField(1, fields[1]), DataField(2, fields[2])}, 2, options)); | |
| ASSERT_OK_AND_ASSIGN(helper, TestHelper::Create(table_path, options, false)); | |
| ASSERT_OK_AND_ASSIGN( | |
| std::unique_ptr<RecordBatch> next_batch, | |
| TestHelper::MakeRecordBatch( | |
| data_type, fmt::format(R"([[{},"{}","{}"]])", kPartitions, lookup_key, payload), | |
| {{"p", fmt::format("{}", kPartitions)}}, kBuckets / 2 - 1, {})); | |
| ASSERT_OK(helper->WriteAndCommit(std::move(next_batch), 1, std::nullopt)); | |
| for (size_t i = 0; i < caches.size(); ++i) { | |
| ASSERT_OK_AND_ASSIGN(ScanResult next, scan(i != 0, caches[i], true)); | |
| ASSERT_EQ(next.cache_hit, 0); | |
| ASSERT_EQ(next.plan->SnapshotId(), 2); | |
| ASSERT_NO_FATAL_FAILURE(check_read(next, kPartitions + 1, true)); | |
| ASSERT_OK_AND_ASSIGN(ScanResult hit, scan(i != 0, caches[i], true)); | |
| ASSERT_EQ(hit.cache_hit, 1); | |
| ASSERT_NO_FATAL_FAILURE(check_read(hit, kPartitions + 1, true)); | |
| } | |
| } | |
| TEST(SelectiveManifestDecodeInteTest, TestSchemaEvolutionPreservesBucketPruning) { | |
| constexpr int32_t kBuckets = 4; | |
| auto key_field = arrow::field("rowkey", arrow::int32()); | |
| auto added_field = arrow::field("value", arrow::int32()); | |
| constexpr int32_t kKey = -1; | |
| auto array = arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({key_field}), | |
| fmt::format("[[{}]]", kKey)) | |
| .ValueOrDie(); | |
| ArrowArray c_array; | |
| ArrowSchema c_schema; | |
| ASSERT_TRUE(arrow::ExportArray(*array, &c_array, &c_schema).ok()); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<BucketIdCalculator> calculator, | |
| BucketIdCalculator::Create(false, kBuckets, GetDefaultPool())); | |
| int32_t bucket = 0; | |
| ASSERT_OK(calculator->CalculateBucketIds(&c_array, &c_schema, &bucket)); | |
| auto dir = UniqueTestDirectory::Create("local"); | |
| ASSERT_TRUE(dir); | |
| std::map<std::string, std::string> options = {{Options::FILE_FORMAT, "orc"}, | |
| {Options::BUCKET, fmt::format("{}", kBuckets)}, | |
| {Options::BUCKET_KEY, "rowkey"}}; | |
| ASSERT_OK_AND_ASSIGN( | |
| std::unique_ptr<TestHelper> helper, | |
| TestHelper::Create(dir->Str(), arrow::schema({key_field}), {}, {}, options, false)); | |
| const std::string table_path = PathUtil::JoinPath(dir->Str(), "foo.db/bar"); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch, | |
| TestHelper::MakeRecordBatch(arrow::struct_({key_field}), | |
| fmt::format("[[{}]]", kKey), {}, bucket, {})); | |
| std::vector<std::unique_ptr<RecordBatch>> batches; | |
| batches.push_back(std::move(batch)); | |
| ASSERT_OK(helper->WriteAndCommit(std::move(batches), 0, std::nullopt)); | |
| helper.reset(); | |
| ASSERT_OK(TestHelper::WriteNextSchema(dir->GetFileSystem(), table_path, | |
| {DataField(0, key_field), DataField(1, added_field)}, 1, | |
| options)); | |
| auto predicate = PredicateBuilder::Equal(0, "rowkey", FieldType::INT, Literal(kKey)); | |
| auto expected = | |
| arrow::ipc::internal::json::ArrayFromJSON( | |
| arrow::struct_({arrow::field("_VALUE_KIND", arrow::int8()), key_field, added_field}), | |
| fmt::format("[[0, {}, null]]", kKey)) | |
| .ValueOrDie(); | |
| for (bool cache_manifest_bytes : {false, true}) { | |
| SCOPED_TRACE(cache_manifest_bytes); | |
| for (bool lazy_decode : {false, true}) { | |
| SCOPED_TRACE(lazy_decode); | |
| std::map<CacheKind, int64_t> capacities = { | |
| {CacheKind::SNAPSHOT_LIVE_MANIFEST, 16 * 1024 * 1024}}; | |
| if (cache_manifest_bytes) { | |
| capacities.emplace(CacheKind::MANIFEST, 16 * 1024 * 1024); | |
| } | |
| auto cache = std::make_shared<CountingRoutingCache>(capacities); | |
| for (int32_t attempt = 0; attempt < 2; ++attempt) { | |
| ScanContextBuilder builder(table_path); | |
| builder.SetPredicate(predicate) | |
| .WithCache(cache) | |
| .AddOption(Options::SCAN_MANIFEST_ENTRY_CACHE_MAX_SNAPSHOTS, "3") | |
| .AddOption(Options::SCAN_MANIFEST_ENTRY_LAZY_DECODE_ENABLED, | |
| lazy_decode ? "true" : "false"); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ScanContext> context, builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableScan> scan, | |
| TableScan::Create(std::move(context))); | |
| ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> plan, scan->CreatePlan()); | |
| ASSERT_EQ(plan->Splits().size(), 1); | |
| auto split = std::dynamic_pointer_cast<DataSplit>(plan->Splits()[0]); | |
| ASSERT_TRUE(split); | |
| ASSERT_EQ(split->Bucket(), bucket); | |
| ASSERT_OK_AND_ASSIGN(uint64_t hit, scan->GetMetrics()->GetCounter( | |
| ScanMetrics::LAST_SNAPSHOT_CACHE_HIT)); | |
| ASSERT_EQ(hit, attempt); | |
| ReadContextBuilder read_builder(table_path); | |
| read_builder.SetPredicate(predicate).EnablePredicateFilter(true); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, | |
| read_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableRead> read, | |
| TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<BatchReader> reader, | |
| read->CreateReader(plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> rows, | |
| ReadResultCollector::CollectResult(std::move(reader))); | |
| ASSERT_TRUE(rows); | |
| ASSERT_TRUE(rows->Equals(std::make_shared<arrow::ChunkedArray>(expected))) | |
| << rows->type()->ToString() << "\n" | |
| << rows->ToString(); | |
| } | |
| } | |
| } | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithAppendBucketKeyPointLookup) { | |
| for (const auto& key_type : {arrow::utf8(), arrow::binary()}) { | |
| SCOPED_TRACE(key_type->ToString()); | |
| auto test_dir = UniqueTestDirectory::Create("local"); | |
| arrow::FieldVector fields = {arrow::field("rowkey", key_type), | |
| arrow::field("value", arrow::int32())}; | |
| std::map<std::string, std::string> options = {{Options::FILE_FORMAT, FileFormat()}, | |
| {Options::FILE_SYSTEM, "local"}, | |
| {Options::BUCKET, "4"}, | |
| {Options::BUCKET_KEY, "rowkey"}}; | |
| ASSERT_OK_AND_ASSIGN( | |
| auto helper, TestHelper::Create(test_dir->Str(), arrow::schema(fields), {}, {}, options, | |
| /*is_streaming_mode=*/false)); | |
| std::string table_path = test_dir->Str() + "/foo.db/bar"; | |
| // Route writes through the public bucket calculator, as a KV client does. | |
| constexpr int32_t kNumKeys = 64; | |
| constexpr int32_t kNumBuckets = 4; | |
| std::vector<std::string> keys; | |
| for (int32_t i = 0; i < kNumKeys; ++i) { | |
| keys.push_back(fmt::format(R"(["key{:03}"])", i)); | |
| } | |
| auto key_array = arrow::ipc::internal::json::ArrayFromJSON( | |
| arrow::struct_({fields[0]}), fmt::format("[{}]", fmt::join(keys, ","))) | |
| .ValueOrDie(); | |
| ::ArrowArray c_keys; | |
| ::ArrowSchema c_schema; | |
| ASSERT_TRUE(arrow::ExportArray(*key_array, &c_keys, &c_schema).ok()); | |
| ASSERT_OK_AND_ASSIGN(auto calculator, | |
| BucketIdCalculator::Create(false, kNumBuckets, GetDefaultPool())); | |
| std::vector<int32_t> bucket_ids(kNumKeys); | |
| ASSERT_OK(calculator->CalculateBucketIds(&c_keys, &c_schema, bucket_ids.data())); | |
| std::vector<std::vector<std::string>> rows(kNumBuckets); | |
| for (int32_t i = 0; i < kNumKeys; ++i) { | |
| rows[bucket_ids[i]].push_back(fmt::format(R"(["key{:03}", {}])", i, i)); | |
| } | |
| std::vector<std::unique_ptr<RecordBatch>> batches; | |
| for (int32_t bucket = 0; bucket < kNumBuckets; ++bucket) { | |
| ASSERT_FALSE(rows[bucket].empty()); | |
| ASSERT_OK_AND_ASSIGN( | |
| auto batch, TestHelper::MakeRecordBatch( | |
| arrow::struct_(fields), | |
| fmt::format("[{}]", fmt::join(rows[bucket], ",")), {}, bucket, {})); | |
| batches.push_back(std::move(batch)); | |
| } | |
| ASSERT_OK(helper->WriteAndCommit(std::move(batches), 0, std::nullopt)); | |
| FieldType field_type = | |
| key_type->id() == arrow::Type::STRING ? FieldType::STRING : FieldType::BINARY; | |
| auto predicate = | |
| PredicateBuilder::Equal(0, "rowkey", field_type, Literal(field_type, "key032", 6)); | |
| // All buckets have overlapping stats for this key. An explicit filter bypasses | |
| // inference, proving that ordinary statistics cannot account for the pruning. | |
| for (int32_t bucket = 0; bucket < kNumBuckets; ++bucket) { | |
| ScanContextBuilder builder(table_path); | |
| builder.SetPredicate(predicate).SetBucketFilter(bucket); | |
| ASSERT_OK_AND_ASSIGN(auto context, FinishScanContext(builder)); | |
| ASSERT_OK_AND_ASSIGN(auto scan, TableScan::Create(std::move(context))); | |
| ASSERT_OK_AND_ASSIGN(auto plan, scan->CreatePlan()); | |
| ASSERT_FALSE(plan->Splits().empty()); | |
| } | |
| int32_t other_index = -1; | |
| for (int32_t i = 0; i < kNumKeys; ++i) { | |
| if (bucket_ids[i] % 2 == bucket_ids[32] % 2 && bucket_ids[i] != bucket_ids[32]) { | |
| other_index = i; | |
| break; | |
| } | |
| } | |
| ASSERT_GE(other_index, 0); | |
| // These keys share a current two-bucket selection but need different historical buckets. | |
| for (int32_t lookup_index : {32, other_index}) { | |
| // The data was written with four buckets. Read it with rescaled table options too. | |
| for (const std::string current_bucket_count : {"2", "4", "8", "17"}) { | |
| SCOPED_TRACE(current_bucket_count); | |
| const std::string lookup_key = fmt::format("key{:03}", lookup_index); | |
| auto lookup_predicate = PredicateBuilder::Equal( | |
| 0, "rowkey", field_type, | |
| Literal(field_type, lookup_key.data(), lookup_key.size())); | |
| ScanContextBuilder scan_builder(table_path); | |
| scan_builder.SetPredicate(lookup_predicate) | |
| .AddOption(Options::BUCKET, current_bucket_count); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto scan, TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(auto plan, scan->CreatePlan()); | |
| if (EnableSnapshotLiveManifestCache()) { | |
| ASSERT_OK_AND_ASSIGN( | |
| uint64_t enabled, | |
| scan->GetMetrics()->GetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_ENABLED)); | |
| ASSERT_EQ(enabled, 1); | |
| ScanContextBuilder cached_builder(table_path); | |
| cached_builder.SetPredicate(lookup_predicate) | |
| .AddOption(Options::BUCKET, current_bucket_count); | |
| ASSERT_OK_AND_ASSIGN(auto cached_context, FinishScanContext(cached_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto cached_scan, | |
| TableScan::Create(std::move(cached_context))); | |
| ASSERT_OK_AND_ASSIGN(plan, cached_scan->CreatePlan()); | |
| ASSERT_OK_AND_ASSIGN(uint64_t hit, cached_scan->GetMetrics()->GetCounter( | |
| ScanMetrics::LAST_SNAPSHOT_CACHE_HIT)); | |
| ASSERT_EQ(hit, 1); | |
| } | |
| ASSERT_FALSE(plan->Splits().empty()); | |
| for (const auto& split : plan->Splits()) { | |
| auto data_split = std::dynamic_pointer_cast<DataSplitImpl>(split); | |
| ASSERT_TRUE(data_split); | |
| ASSERT_EQ(data_split->Bucket(), bucket_ids[lookup_index]); | |
| } | |
| ReadContextBuilder read_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_builder); | |
| read_builder.SetPredicate(lookup_predicate).EnablePredicateFilter(true); | |
| ASSERT_OK_AND_ASSIGN(auto read_context, read_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto reader, read->CreateReader(plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(auto result, | |
| ReadResultCollector::CollectResult(std::move(reader))); | |
| auto expected_fields = fields; | |
| expected_fields.insert(expected_fields.begin(), | |
| arrow::field("_VALUE_KIND", arrow::int8())); | |
| auto expected_array = | |
| arrow::ipc::internal::json::ArrayFromJSON( | |
| arrow::struct_(expected_fields), | |
| fmt::format(R"([[0, "{}", {}]])", lookup_key, lookup_index)) | |
| .ValueOrDie(); | |
| auto expected = std::make_shared<arrow::ChunkedArray>(expected_array); | |
| ASSERT_TRUE(expected->Equals(result)) << result->ToString(); | |
| } | |
| } | |
| } | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithAppendSnapshot5) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = GetDataDir() + "/" + file_format + "/append_09.db/append_09"; | |
| // scan | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "5"); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 5); | |
| auto splits = result_plan->Splits(); | |
| ASSERT_EQ(3, splits.size()); | |
| // read | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(splits)); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Alice", 10, 1, 11.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Tony", 10, 0, 14.1], | |
| [0, "Emily", 10, 0, 15.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "Alex", 10, 0, 16.1], | |
| [0, "David", 10, 0, 17.1], | |
| [0, "Lily", 10, 0, 17.1], | |
| [0, "Lucy", 20, 1, 14.1], | |
| [0, "Paul", 20, 1, null] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| ASSERT_OK_AND_ASSIGN(auto count_reader, table_read->CreateCountReader(splits)); | |
| ASSERT_OK_AND_ASSIGN(int64_t count, count_reader->CountRows()); | |
| ASSERT_EQ(count, read_result->length()); | |
| } | |
| // One reader is built per data file of a split, and those readers are built in parallel. | |
| // Snapshot 5 has a bucket holding five files, so the parallel build path is exercised here and | |
| // has to return the very same rows in the very same order as the other snapshot-5 read above. | |
| TEST_P(ScanAndReadInteTest, TestWithAppendSnapshot5WithReaderBuildParallelism) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = GetDataDir() + "/" + file_format + "/append_09.db/append_09"; | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "5"); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| auto splits = result_plan->Splits(); | |
| ASSERT_EQ(3, splits.size()); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(splits)); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // The parallel build has to return exactly the rows of the sequential snapshot-5 read above, | |
| // in the very same order. | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Alice", 10, 1, 11.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Tony", 10, 0, 14.1], | |
| [0, "Emily", 10, 0, 15.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "Alex", 10, 0, 16.1], | |
| [0, "David", 10, 0, 17.1], | |
| [0, "Lily", 10, 0, 17.1], | |
| [0, "Lucy", 20, 1, 14.1], | |
| [0, "Paul", 20, 1, null] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithAppendSnapshotWithStreamWithDefaultMode) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = GetDataDir() + "/" + file_format + "/append_09.db/append_09"; | |
| // scan | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "1").WithStreamingMode(true); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| std::vector<std::optional<int64_t>> expected_snapshot_ids = {std::nullopt, 1, 2, 3, 4}; | |
| auto expected_snapshot1 = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Alice", 10, 1, 11.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Tony", 10, 0, 14.1], | |
| [0, "Lucy", 20, 1, 14.1] | |
| ])") | |
| .ValueOrDie()); | |
| auto expected_snapshot2 = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Emily", 10, 0, 15.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "Alex", 10, 0, 16.1], | |
| [0, "Paul", 20, 1, null] | |
| ])") | |
| .ValueOrDie()); | |
| auto expected_snapshot3 = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "David", 10, 0, 17.1] | |
| ])") | |
| .ValueOrDie()); | |
| auto expected_snapshot4 = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Lily", 10, 0, 17.1] | |
| ])") | |
| .ValueOrDie()); | |
| std::vector<std::shared_ptr<arrow::ChunkedArray>> expected_arrays = { | |
| nullptr, expected_snapshot1, expected_snapshot2, expected_snapshot3, expected_snapshot4}; | |
| CheckStreamScanResult(table_scan, table_read, expected_snapshot_ids, expected_arrays); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestJavaPaimon1WithAppendSnapshot1) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = GetDataDir() + "/" + file_format + "/append_10.db/append_10"; | |
| // scan | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "1"); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 1); | |
| auto splits = result_plan->Splits(); | |
| ASSERT_EQ(3, splits.size()); | |
| // read | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(splits)); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Alice", 10, 1, 11.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Tony", 10, 0, 14.1], | |
| [0, "Lucy", 20, 1, 14.1] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestJavaPaimon1WithAppendSnapshotOfNestedType) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = GetDataDir() + "/" + file_format + | |
| "/append_complex_build_in_fieldid.db/" | |
| "append_complex_build_in_fieldid/"; | |
| // scan | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "1"); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 1); | |
| auto splits = result_plan->Splits(); | |
| ASSERT_EQ(1, splits.size()); | |
| // read | |
| auto map_type = arrow::map(arrow::int8(), arrow::int16()); | |
| auto list_type = arrow::list(DataField::ConvertDataFieldToArrowField( | |
| DataField(536871936, arrow::field("item", arrow::float32())))); | |
| std::vector<DataField> struct_fields = {DataField(3, arrow::field("f0", arrow::boolean())), | |
| DataField(4, arrow::field("f1", arrow::int64()))}; | |
| auto struct_type = DataField::ConvertDataFieldsToArrowStructType(struct_fields); | |
| std::vector<DataField> read_fields = { | |
| SpecialFields::ValueKind(), | |
| DataField(0, arrow::field("f1", map_type)), | |
| DataField(1, arrow::field("f2", list_type)), | |
| DataField(2, arrow::field("f3", struct_type)), | |
| DataField(5, arrow::field("f4", arrow::timestamp(arrow::TimeUnit::NANO))), | |
| DataField(6, arrow::field("f5", arrow::date32())), | |
| DataField(7, arrow::field("f6", arrow::decimal128(2, 2)))}; | |
| std::shared_ptr<arrow::DataType> arrow_data_type = | |
| DataField::ConvertDataFieldsToArrowStructType(read_fields); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(splits)); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result | |
| std::shared_ptr<arrow::ChunkedArray> expected_array; | |
| auto array_status = arrow::ipc::internal::json::ChunkedArrayFromJSON(arrow_data_type, {R"([ | |
| [0, [[0, 0]], [0.1, 0.2], [true, 2], "1970-01-01 00:02:03.123123", 2456, "0.22"], | |
| [0, [[127, 32767], [-128, -32768]], [1.1, 1.2], [false, 2222], "1970-01-01 00:02:03.123123", 245, "0.12"], | |
| [0, [[1, 64], [2, 32]], [2.2, 3.2], [true, 2], "1970-01-01 00:00:00.0", 24, null] | |
| ])"}, | |
| &expected_array); | |
| ASSERT_TRUE(array_status.ok()); | |
| ASSERT_TRUE(expected_array->Equals(read_result)) << read_result->ToString(); | |
| } | |
| // test pk with dv | |
| TEST_P(ScanAndReadInteTest, TestWithPKWithDvBatchScanSnapshot6) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = GetDataDir() + "/" + file_format + | |
| "/pk_table_scan_and_read_dv.db/pk_table_scan_and_read_dv/"; | |
| // normal batch scan case for pk+dv, all data in level 0 is filtered out | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "6"); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| ASSERT_OK_AND_ASSIGN(auto read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 6); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(result_plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Two roads diverged in a wood, and I took the one less traveled by, And that has made all the difference.", 10, 1, 11.0], | |
| [0, "Alice", 10, 1, 19.1], | |
| [0, "Alex", 10, 0, 16.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "David", 10, 0, 17.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Lucy", 20, 1, 14.1], | |
| [0, "Paul", 20, 1, 18.1] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| // CountRows should match the number of visible rows returned by CreateReader. | |
| ASSERT_OK_AND_ASSIGN(auto count_reader, table_read->CreateCountReader(result_plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(int64_t count, count_reader->CountRows()); | |
| ASSERT_EQ(count, read_result->length()); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithPKWithDvBatchScanSnapshot1) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = GetDataDir() + "/" + file_format + | |
| "/pk_table_scan_and_read_dv.db/pk_table_scan_and_read_dv/"; | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "1"); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(auto read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 1); | |
| // snapshot 1 is an append snapshot without compact, data is all in level 0, therefore batch | |
| // scan return empty plan | |
| ASSERT_TRUE(result_plan->Splits().empty()); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithPKWithDvBatchScanSnapshot6WithPartitionAndBucketFilter) { | |
| auto file_format = FileFormat(); | |
| // all data in level 0 & not in partition 10, bucket 1 is filtered out | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_dv.db/pk_table_scan_and_read_dv/"; | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "6"); | |
| scan_context_builder.SetBucketFilter(1).SetPartitionFilter({{{"f1", "10"}}}); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(auto read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 6); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(result_plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Alex", 10, 0, 16.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "David", 10, 0, 17.1], | |
| [0, "Emily", 10, 0, 13.1] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithPKWithDvBatchScanSnapshot6WithPredicate) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_dv.db/pk_table_scan_and_read_dv/"; | |
| // predicate: f0 != "Alice" (key predicate) and f3 > 18 (value predicate) and all data in level | |
| // 0 | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "6"); | |
| std::string literal_str = "Alice"; | |
| auto not_equal = PredicateBuilder::NotEqual( | |
| /*field_index=*/0, /*field_name=*/"f0", FieldType::STRING, | |
| Literal(FieldType::STRING, literal_str.data(), literal_str.size())); | |
| auto greater_than = PredicateBuilder::GreaterThan(/*field_index=*/3, /*field_name=*/"f3", | |
| FieldType::DOUBLE, Literal(18.0)); | |
| ASSERT_OK_AND_ASSIGN(auto predicate, PredicateBuilder::And({not_equal, greater_than})); | |
| scan_context_builder.SetPredicate(predicate); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| read_context_builder.SetPredicate(predicate).EnableLateMaterializing(false); | |
| ASSERT_OK_AND_ASSIGN(auto read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 6); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(result_plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Lucy", 20, 1, 14.1], | |
| [0, "Paul", 20, 1, 18.1] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithPKWithDvBatchScanSnapshot6WithLateMaterializing) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_dv.db/pk_table_scan_and_read_dv/"; | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "6"); | |
| std::string literal_str = "Alice"; | |
| auto not_equal = PredicateBuilder::NotEqual( | |
| /*field_index=*/0, /*field_name=*/"f0", FieldType::STRING, | |
| Literal(FieldType::STRING, literal_str.data(), literal_str.size())); | |
| auto greater_than = PredicateBuilder::GreaterThan(/*field_index=*/3, /*field_name=*/"f3", | |
| FieldType::DOUBLE, Literal(18.0)); | |
| ASSERT_OK_AND_ASSIGN(auto predicate, PredicateBuilder::And({not_equal, greater_than})); | |
| scan_context_builder.SetPredicate(predicate); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| read_context_builder.SetPredicate(predicate).EnableLateMaterializing(true); | |
| ASSERT_OK_AND_ASSIGN(auto read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 6); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(result_plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result: "Lucy" (f3 = 14.1) does not match f3 > 18 and is filtered out. | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Paul", 20, 1, 18.1] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithPKWithDvBatchScanSnapshot4WithPredicate) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_dv.db/pk_table_scan_and_read_dv/"; | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "4"); | |
| scan_context_builder.SetBucketFilter(0).SetPartitionFilter({{{"f1", "10"}}}); | |
| // predicate f3 > 20.0 will be applied, as in kv mode value filter is enabled | |
| auto predicate = PredicateBuilder::GreaterThan(/*field_index=*/3, /*field_name=*/"f3", | |
| FieldType::DOUBLE, Literal(20.0)); | |
| scan_context_builder.SetPredicate(predicate); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 4); | |
| ASSERT_TRUE(result_plan->Splits().empty()); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithPKWithDvBatchScanSnapshot6WithLimit) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_dv.db/pk_table_scan_and_read_dv/"; | |
| // normal batch scan case for pk+dv, all data in level 0 is filtered out, set row limits to 6, | |
| // data in partition 20 is truncated | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.SetLimit(6); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(auto read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 6); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(result_plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Two roads diverged in a wood, and I took the one less traveled by, And that has made all the difference.", 10, 1, 11.0], | |
| [0, "Alice", 10, 1, 19.1], | |
| [0, "Alex", 10, 0, 16.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "David", 10, 0, 17.1], | |
| [0, "Emily", 10, 0, 13.1] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithPKWithDvStreamFromSnapshot4) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_dv.db/pk_table_scan_and_read_dv/"; | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "4") | |
| .AddOption(Options::SCAN_MODE, "from-snapshot-full") | |
| .WithStreamingMode(true); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| // from an compact snapshot | |
| std::vector<std::optional<int64_t>> expected_snapshot_ids = {4, 5, 6}; | |
| auto expected_snapshot4_batch = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Two roads diverged in a wood, and I took the one less traveled by, And that has made all the difference.", 10, 1, 11.0], | |
| [0, "Alice", 10, 1, 19.1], | |
| [0, "Alex", 10, 0, 16.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "David", 10, 0, 17.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Lucy", 20, 1, 14.1], | |
| [0, "Paul", 20, 1, 18.1] | |
| ])") | |
| .ValueOrDie()); | |
| auto expected_snapshot5_stream = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [2, "Alex", 10, 0, 21.2], | |
| [0, "Marco", 10, 0, 21.1], | |
| [0, "Skye", 10, 0, 21.0] | |
| ])") | |
| .ValueOrDie()); | |
| auto expected_snapshot6_stream = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [3, "Alex", 10, 0, 31.2], | |
| [0, "Marco2", 10, 0, 31.1], | |
| [0, "Skye2", 10, 0, 31.0] | |
| ])") | |
| .ValueOrDie()); | |
| std::vector<std::shared_ptr<arrow::ChunkedArray>> expected_arrays = { | |
| expected_snapshot4_batch, expected_snapshot5_stream, expected_snapshot6_stream}; | |
| CheckStreamScanResult(table_scan, table_read, expected_snapshot_ids, expected_arrays); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithPKWithDvStreamFromSnapshot5) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_dv.db/pk_table_scan_and_read_dv/"; | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "5") | |
| .AddOption(Options::SCAN_MODE, "from-snapshot-full") | |
| .WithStreamingMode(true); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| // from an append snapshot, first plan is snapshot5 with merge, second plan is snapshot6 | |
| // (isStreaming=true) | |
| std::vector<std::optional<int64_t>> expected_snapshot_ids = {5, 6}; | |
| auto expected_snapshot5_batch = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Two roads diverged in a wood, and I took the one less traveled by, And that has made all the difference.", 10, 1, 11.0], | |
| [0, "Alice", 10, 1, 19.1], | |
| [0, "Alex", 10, 0, 21.2], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "David", 10, 0, 17.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Marco", 10, 0, 21.1], | |
| [0, "Skye", 10, 0, 21.0], | |
| [0, "Lucy", 20, 1, 14.1], | |
| [0, "Paul", 20, 1, 18.1] | |
| ])") | |
| .ValueOrDie()); | |
| auto expected_snapshot6_stream = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [3, "Alex", 10, 0, 31.2], | |
| [0, "Marco2", 10, 0, 31.1], | |
| [0, "Skye2", 10, 0, 31.0] | |
| ])") | |
| .ValueOrDie()); | |
| std::vector<std::shared_ptr<arrow::ChunkedArray>> expected_arrays = {expected_snapshot5_batch, | |
| expected_snapshot6_stream}; | |
| CheckStreamScanResult(table_scan, table_read, expected_snapshot_ids, expected_arrays); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithPKWithDvStreamFromSnapshot6) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_dv.db/pk_table_scan_and_read_dv/"; | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_MODE, "latest-full").WithStreamingMode(true); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| // from an append snapshot, first plan is snapshot6 with merge, level 0 data in base manifest | |
| // list is contained | |
| std::vector<std::optional<int64_t>> expected_snapshot_ids = {6}; | |
| auto expected_snapshot6_batch = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Two roads diverged in a wood, and I took the one less traveled by, And that has made all the difference.", 10, 1, 11.0], | |
| [0, "Alice", 10, 1, 19.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "David", 10, 0, 17.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Marco", 10, 0, 21.1], | |
| [0, "Marco2", 10, 0, 31.1], | |
| [0, "Skye", 10, 0, 21.0], | |
| [0, "Skye2", 10, 0, 31.0], | |
| [0, "Lucy", 20, 1, 14.1], | |
| [0, "Paul", 20, 1, 18.1] | |
| ])") | |
| .ValueOrDie()); | |
| std::vector<std::shared_ptr<arrow::ChunkedArray>> expected_arrays = {expected_snapshot6_batch}; | |
| CheckStreamScanResult(table_scan, table_read, expected_snapshot_ids, expected_arrays); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithPKWithDvStreamFromSnapshot1) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_dv.db/pk_table_scan_and_read_dv/"; | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "1") | |
| .AddOption(Options::SCAN_MODE, "from-snapshot-full") | |
| .WithStreamingMode(true); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| // from the first snapshot | |
| std::vector<std::optional<int64_t>> expected_snapshot_ids = {1, 3, 5, 6}; | |
| auto expected_snapshot1_batch = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Alice", 10, 1, 11.1], | |
| [0, "Two roads diverged in a wood, and I took the one less traveled by, And that has made all the difference.", 10, 1, 11.0], | |
| [0, "Alex", 10, 0, 16.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "David", 10, 0, 17.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Tony", 10, 0, 14.1], | |
| [0, "Lucy", 20, 1, 14.1], | |
| [0, "Paul", 20, 1, 18.1] | |
| ])") | |
| .ValueOrDie()); | |
| auto expected_snapshot3_stream = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [2, "Alice", 10, 1, 19.1], | |
| [3, "Tony", 10, 0, 14.1] | |
| ])") | |
| .ValueOrDie()); | |
| auto expected_snapshot5_stream = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [2, "Alex", 10, 0, 21.2], | |
| [0, "Marco", 10, 0, 21.1], | |
| [0, "Skye", 10, 0, 21.0] | |
| ])") | |
| .ValueOrDie()); | |
| auto expected_snapshot6_stream = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [3, "Alex", 10, 0, 31.2], | |
| [0, "Marco2", 10, 0, 31.1], | |
| [0, "Skye2", 10, 0, 31.0] | |
| ])") | |
| .ValueOrDie()); | |
| std::vector<std::shared_ptr<arrow::ChunkedArray>> expected_arrays = { | |
| expected_snapshot1_batch, expected_snapshot3_stream, expected_snapshot5_stream, | |
| expected_snapshot6_stream}; | |
| CheckStreamScanResult(table_scan, table_read, expected_snapshot_ids, expected_arrays); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithPKWithDvStreamFromSnapshot2) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_dv.db/pk_table_scan_and_read_dv/"; | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "2") | |
| .AddOption(Options::SCAN_MODE, "from-snapshot") | |
| .WithStreamingMode(true); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| // use from-snapshot mode, first plan is empty, second plan is snapshot 3 with delta files | |
| // (snapshot 2 is compact snapshot, so skipped) | |
| std::vector<std::optional<int64_t>> expected_snapshot_ids = {std::nullopt, 3, 5, 6}; | |
| auto expected_snapshot3_stream = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [2, "Alice", 10, 1, 19.1], | |
| [3, "Tony", 10, 0, 14.1] | |
| ])") | |
| .ValueOrDie()); | |
| auto expected_snapshot5_stream = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [2, "Alex", 10, 0, 21.2], | |
| [0, "Marco", 10, 0, 21.1], | |
| [0, "Skye", 10, 0, 21.0] | |
| ])") | |
| .ValueOrDie()); | |
| auto expected_snapshot6_stream = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [3, "Alex", 10, 0, 31.2], | |
| [0, "Marco2", 10, 0, 31.1], | |
| [0, "Skye2", 10, 0, 31.0] | |
| ])") | |
| .ValueOrDie()); | |
| std::vector<std::shared_ptr<arrow::ChunkedArray>> expected_arrays = { | |
| nullptr, expected_snapshot3_stream, expected_snapshot5_stream, expected_snapshot6_stream}; | |
| CheckStreamScanResult(table_scan, table_read, expected_snapshot_ids, expected_arrays); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithPKWithNestedType) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = | |
| paimon::test::GetDataDir() + file_format + "/pk_table_nested_type.db/pk_table_nested_type/"; | |
| ScanContextBuilder scan_context_builder(table_path); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN( | |
| auto read_context, | |
| read_context_builder.SetReadFieldNames({"shopId", "dt", "hr", "col0", "col1", "col2"}) | |
| .Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 2); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(result_plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| auto struct_inner_type = | |
| arrow::struct_({arrow::field("f0", arrow::utf8()), arrow::field("f1", arrow::int32()), | |
| arrow::field("f2", arrow::list(arrow::int32()))}); | |
| auto map_type = arrow::map(arrow::int32(), arrow::int32()); | |
| auto data_type = | |
| arrow::struct_({arrow::field("_VALUE_KIND", arrow::int8()), | |
| arrow::field("shopId", arrow::int32()), arrow::field("dt", arrow::utf8()), | |
| arrow::field("hr", arrow::int32()), arrow::field("col0", arrow::int32()), | |
| arrow::field("col1", struct_inner_type), arrow::field("col2", map_type)}); | |
| auto array1 = arrow::ipc::internal::json::ArrayFromJSON(data_type, R"([ | |
| [0, 1005, "2025-04-14", 10, 50, ["str-5", 500, [5, null, 6]], [[5, 5], [105, 105]]], | |
| [0, 1006, "2025-04-14", 10, 60, ["str-6", 600, [6, null, 7]], [[6, 6], [106, 106]]] | |
| ])") | |
| .ValueOrDie(); | |
| auto array2 = arrow::ipc::internal::json::ArrayFromJSON(data_type, R"([ | |
| [0, 1001, "2025-04-14", 10, 10, ["str-1", 100, [1, null, 2]], [[1, 1], [101, 101]]], | |
| [0, 1004, "2025-04-14", 10, 40, ["str-4", 400, [4, null, 5]], [[4, 4], [104, 104]]], | |
| [0, 1007, "2025-04-14", 10, 70, ["str-7", 700, [7, null, 8]], [[7, 7], [107, 107]]] | |
| ])") | |
| .ValueOrDie(); | |
| auto expected_array = arrow::ChunkedArray::Make({array1, array2}).ValueOrDie(); | |
| ASSERT_TRUE(expected_array->Equals(read_result)) << read_result->ToString(); | |
| } | |
| // test pk with mor | |
| TEST_P(ScanAndReadInteTest, TestWithPKWithMorBatchScanLatestSnapshot) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_mor.db/pk_table_scan_and_read_mor/"; | |
| // normal batch scan case for pk+mor, use latest snapshot if not specified | |
| ScanContextBuilder scan_context_builder(table_path); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(auto read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 5); | |
| auto splits = result_plan->Splits(); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(splits)); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "David", 10, 0, 17.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Marco", 10, 0, 21.1], | |
| [0, "Marco2", 10, 0, 31.1], | |
| [0, "Skye", 10, 0, 21.0], | |
| [0, "Skye2", 10, 0, 31.0], | |
| [0, "Alice", 10, 1, 19.1], | |
| [0, "Two roads diverged in a wood, and I took the one less traveled by, And that has made all the difference.", 10, 1, 11.0], | |
| [0, "Lucy", 20, 1, 14.1], | |
| [0, "Paul", 20, 1, 18.1] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| ASSERT_OK_AND_ASSIGN(auto count_reader, table_read->CreateCountReader(splits)); | |
| ASSERT_OK_AND_ASSIGN(int64_t count, count_reader->CountRows()); | |
| ASSERT_EQ(count, read_result->length()); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithPKWithMorBatchScanSnapshot2) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_mor.db/pk_table_scan_and_read_mor/"; | |
| // normal batch scan case for pk+mor, read snapshot 2 (append snapshot), all data is in level 0 | |
| // with merge read | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "2"); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(auto read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 2); | |
| auto splits = result_plan->Splits(); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(splits)); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Alice", 10, 1, 19.1], | |
| [0, "Two roads diverged in a wood, and I took the one less traveled by, And that has made all the difference.", 10, 1, 11.0], | |
| [0, "Alex", 10, 0, 16.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "David", 10, 0, 17.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Lucy", 20, 1, 14.1], | |
| [0, "Paul", 20, 1, 18.1] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| ASSERT_OK_AND_ASSIGN(auto count_reader, table_read->CreateCountReader(splits)); | |
| ASSERT_OK_AND_ASSIGN(int64_t count, count_reader->CountRows()); | |
| ASSERT_EQ(count, read_result->length()); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithPKWithMorBatchScanSnapshot5WithPartitionAndBucketFilter) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_mor.db/pk_table_scan_and_read_mor/"; | |
| // all data not in partition 10, bucket 1 is filtered out | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "5"); | |
| scan_context_builder.SetBucketFilter(1).SetPartitionFilter({{{"f1", "10"}}}); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(auto read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 5); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(result_plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "David", 10, 0, 17.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Marco", 10, 0, 21.1], | |
| [0, "Marco2", 10, 0, 31.1], | |
| [0, "Skye", 10, 0, 21.0], | |
| [0, "Skye2", 10, 0, 31.0] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithPKWithMorBatchScanSnapshot5WithPredicate) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_mor.db/pk_table_scan_and_read_mor/"; | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "5"); | |
| // predicate: f0 != "Alice" and f0 < "Lucy" and f3 <= 30.0 | |
| // as f3 <= 30.0 is a value filter, it does not take effect | |
| // f0 < "Lucy" will skip the data file which delete "Tony" in scan. | |
| // Therefore, the false positives returned by the pk table for predicate pushdown may be | |
| // incorrect. (e.g., Tony is deleted but returned as a insert record) | |
| std::string literal_str = "Alice"; | |
| auto not_equal = PredicateBuilder::NotEqual( | |
| /*field_index=*/0, /*field_name=*/"f0", FieldType::STRING, | |
| Literal(FieldType::STRING, literal_str.data(), literal_str.size())); | |
| std::string literal_str2 = "Lucy"; | |
| auto less_than = PredicateBuilder::LessThan( | |
| /*field_index=*/0, /*field_name=*/"f0", FieldType::STRING, | |
| Literal(FieldType::STRING, literal_str2.data(), literal_str2.size())); | |
| auto less_or_equal = PredicateBuilder::LessOrEqual(/*field_index=*/3, /*field_name=*/"f3", | |
| FieldType::DOUBLE, Literal(30.0)); | |
| ASSERT_OK_AND_ASSIGN(auto predicate, | |
| PredicateBuilder::And({not_equal, less_than, less_or_equal})); | |
| scan_context_builder.SetPredicate(predicate); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| read_context_builder.SetPredicate(predicate).EnableLateMaterializing(false); | |
| ASSERT_OK_AND_ASSIGN(auto read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 5); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(result_plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "David", 10, 0, 17.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Marco", 10, 0, 21.1], | |
| [0, "Marco2", 10, 0, 31.1], | |
| [0, "Skye", 10, 0, 21.0], | |
| [0, "Skye2", 10, 0, 31.0], | |
| [0, "Tony", 10, 0, 14.1], | |
| [0, "Alice", 10, 1, 19.1], | |
| [0, "Two roads diverged in a wood, and I took the one less traveled by, And that has made all the difference.", 10, 1, 11.0] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| } | |
| // Same coverage as the deletion-vector case above, for the merge-on-read path where only the | |
| // key part of the predicate is pushed down into the data files. | |
| TEST_P(ScanAndReadInteTest, TestWithPKWithMorBatchScanSnapshot5WithLateMaterializing) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_mor.db/pk_table_scan_and_read_mor/"; | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "5"); | |
| std::string literal_str = "Alice"; | |
| auto not_equal = PredicateBuilder::NotEqual( | |
| /*field_index=*/0, /*field_name=*/"f0", FieldType::STRING, | |
| Literal(FieldType::STRING, literal_str.data(), literal_str.size())); | |
| std::string literal_str2 = "Lucy"; | |
| auto less_than = PredicateBuilder::LessThan( | |
| /*field_index=*/0, /*field_name=*/"f0", FieldType::STRING, | |
| Literal(FieldType::STRING, literal_str2.data(), literal_str2.size())); | |
| auto less_or_equal = PredicateBuilder::LessOrEqual(/*field_index=*/3, /*field_name=*/"f3", | |
| FieldType::DOUBLE, Literal(30.0)); | |
| ASSERT_OK_AND_ASSIGN(auto predicate, | |
| PredicateBuilder::And({not_equal, less_than, less_or_equal})); | |
| scan_context_builder.SetPredicate(predicate); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| read_context_builder.SetPredicate(predicate).EnableLateMaterializing(true); | |
| ASSERT_OK_AND_ASSIGN(auto read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 5); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(result_plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result: only the rows before "Lucy" with f3 <= 30.0 remain. | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "David", 10, 0, 17.1], | |
| [0, "Emily", 10, 0, 13.1] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithPKWithMorBatchScanSnapshot3WithPredicate) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_mor.db/pk_table_scan_and_read_mor/"; | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "3"); | |
| scan_context_builder.SetBucketFilter(1).SetPartitionFilter({{{"f1", "10"}}}); | |
| // predicate f3 > 20.0 will be applied as all data files in bucket 1 in compact snapshot 3 are | |
| // all filtered | |
| auto predicate = PredicateBuilder::GreaterThan(/*field_index=*/3, /*field_name=*/"f3", | |
| FieldType::DOUBLE, Literal(20.0)); | |
| scan_context_builder.SetPredicate(predicate); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 3); | |
| ASSERT_TRUE(result_plan->Splits().empty()); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithPKWithDvWithInvalidAggregateBatchScanSnapshot3) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_dv.db/pk_table_scan_and_read_dv/"; | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "3") | |
| .AddOption(Options::MERGE_ENGINE, "aggregation") | |
| .AddOption("fields.f3.aggregate-function", "rbm32"); | |
| scan_context_builder.SetBucketFilter(1).SetPartitionFilter({{{"f1", "10"}}}); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| read_context_builder.AddOption(Options::MERGE_ENGINE, "aggregation") | |
| .AddOption("fields.f3.aggregate-function", "rbm32"); | |
| ASSERT_OK_AND_ASSIGN(auto read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 3); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(result_plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Alex", 10, 0, 16.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "David", 10, 0, 17.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Tony", 10, 0, 14.1] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithPKWithMorWithInvalidAggregateBatchScanSnapshot3) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_mor.db/pk_table_scan_and_read_mor/"; | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "3") | |
| .AddOption(Options::MERGE_ENGINE, "aggregation") | |
| .AddOption("fields.f3.aggregate-function", "rbm32"); | |
| scan_context_builder.SetBucketFilter(1).SetPartitionFilter({{{"f1", "10"}}}); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| read_context_builder.AddOption(Options::MERGE_ENGINE, "aggregation") | |
| .AddOption("fields.f3.aggregate-function", "rbm32"); | |
| ASSERT_OK_AND_ASSIGN(auto read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 3); | |
| ASSERT_NOK_WITH_MSG( | |
| table_read->CreateReader(result_plan->Splits()), | |
| "Use unsupported aggregation rbm32 or spell aggregate function incorrectly"); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithPKWithAggregateBatchScanSnapshot3WithPredicate) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_mor.db/pk_table_scan_and_read_mor/"; | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "3") | |
| .AddOption(Options::MERGE_ENGINE, "aggregation") | |
| .AddOption("fields.f3.aggregate-function", "sum"); | |
| scan_context_builder.SetBucketFilter(1).SetPartitionFilter({{{"f1", "10"}}}); | |
| // predicate f3 > 20.0 will not be applied, as merge engine is aggregation | |
| auto predicate = PredicateBuilder::GreaterThan(/*field_index=*/3, /*field_name=*/"f3", | |
| FieldType::DOUBLE, Literal(20.0)); | |
| scan_context_builder.SetPredicate(predicate); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| read_context_builder.AddOption(Options::MERGE_ENGINE, "aggregation") | |
| .AddOption("fields.f3.aggregate-function", "sum"); | |
| ASSERT_OK_AND_ASSIGN(auto read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 3); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(result_plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Alex", 10, 0, 16.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "David", 10, 0, 17.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Tony", 10, 0, 0] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithPKWithPartialUpdateBatchScanSnapshot3WithPredicate) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_mor.db/pk_table_scan_and_read_mor/"; | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "3") | |
| .AddOption(Options::MERGE_ENGINE, "partial-update") | |
| .AddOption(Options::IGNORE_DELETE, "true"); | |
| scan_context_builder.SetBucketFilter(1).SetPartitionFilter({{{"f1", "10"}}}); | |
| // predicate f3 > 20.0 will not be applied, as merge engine is partial-update | |
| auto predicate = PredicateBuilder::GreaterThan(/*field_index=*/3, /*field_name=*/"f3", | |
| FieldType::DOUBLE, Literal(20.0)); | |
| scan_context_builder.SetPredicate(predicate); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| read_context_builder.AddOption(Options::MERGE_ENGINE, "partial-update") | |
| .AddOption(Options::IGNORE_DELETE, "true"); | |
| ASSERT_OK_AND_ASSIGN(auto read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 3); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(result_plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| ASSERT_TRUE(read_result); | |
| // check result | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Alex", 10, 0, 16.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "David", 10, 0, 17.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Tony", 10, 0, 14.1] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithPKWithMorBatchScanSnapshot5WithLimit) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_mor.db/pk_table_scan_and_read_mor/"; | |
| // in pk mor mode, limit does not take effect, as we do not know the number of records after | |
| // merging | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.SetLimit(6); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(auto read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 5); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(result_plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "David", 10, 0, 17.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Marco", 10, 0, 21.1], | |
| [0, "Marco2", 10, 0, 31.1], | |
| [0, "Skye", 10, 0, 21.0], | |
| [0, "Skye2", 10, 0, 31.0], | |
| [0, "Alice", 10, 1, 19.1], | |
| [0, "Two roads diverged in a wood, and I took the one less traveled by, And that has made all the difference.", 10, 1, 11.0], | |
| [0, "Lucy", 20, 1, 14.1], [0, "Paul", 20, 1, 18.1] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithPKWithMorStreamFromSnapshot4) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_mor.db/pk_table_scan_and_read_mor/"; | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "4") | |
| .AddOption(Options::SCAN_MODE, "from-snapshot-full") | |
| .WithStreamingMode(true); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| // first plan is snapshot4 (isStreaming=false), second plan is snapshot5 | |
| std::vector<std::optional<int64_t>> expected_snapshot_ids = {4, 5}; | |
| auto expected_snapshot4_batch = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Alex", 10, 0, 21.2], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "David", 10, 0, 17.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Marco", 10, 0, 21.1], | |
| [0, "Skye", 10, 0, 21.0], | |
| [0, "Alice", 10, 1, 19.1], | |
| [0, "Two roads diverged in a wood, and I took the one less traveled by, And that has made all the difference.", 10, 1, 11.0], | |
| [0, "Lucy", 20, 1, 14.1], | |
| [0, "Paul", 20, 1, 18.1] | |
| ])") | |
| .ValueOrDie()); | |
| auto expected_snapshot5_stream = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [3, "Alex", 10, 0, 31.2], | |
| [0, "Marco2", 10, 0, 31.1], | |
| [0, "Skye2", 10, 0, 31.0] | |
| ])") | |
| .ValueOrDie()); | |
| std::vector<std::shared_ptr<arrow::ChunkedArray>> expected_arrays = {expected_snapshot4_batch, | |
| expected_snapshot5_stream}; | |
| CheckStreamScanResult(table_scan, table_read, expected_snapshot_ids, expected_arrays); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithPKWithMorStreamFromSnapshot1) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_mor.db/pk_table_scan_and_read_mor/"; | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "1") | |
| .AddOption(Options::SCAN_MODE, "from-snapshot-full") | |
| .WithStreamingMode(true); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| std::vector<std::optional<int64_t>> expected_snapshot_ids = {1, 2, 4, 5}; | |
| auto expected_snapshot1_batch = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Alice", 10, 1, 11.1], | |
| [0, "Two roads diverged in a wood, and I took the one less traveled by, And that has made all the difference.", 10, 1, 11.0], | |
| [0, "Alex", 10, 0, 16.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "David", 10, 0, 17.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Tony", 10, 0, 14.1], | |
| [0, "Lucy", 20, 1, 14.1], | |
| [0, "Paul", 20, 1, 18.1] | |
| ])") | |
| .ValueOrDie()); | |
| auto expected_snapshot2_stream = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [2, "Alice", 10, 1, 19.1], | |
| [3, "Tony", 10, 0, 14.1] | |
| ])") | |
| .ValueOrDie()); | |
| auto expected_snapshot4_stream = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [2, "Alex", 10, 0, 21.2], | |
| [0, "Marco", 10, 0, 21.1], | |
| [0, "Skye", 10, 0, 21.0] | |
| ])") | |
| .ValueOrDie()); | |
| auto expected_snapshot5_stream = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [3, "Alex", 10, 0, 31.2], | |
| [0, "Marco2", 10, 0, 31.1], | |
| [0, "Skye2", 10, 0, 31.0] | |
| ])") | |
| .ValueOrDie()); | |
| std::vector<std::shared_ptr<arrow::ChunkedArray>> expected_arrays = { | |
| expected_snapshot1_batch, expected_snapshot2_stream, expected_snapshot4_stream, | |
| expected_snapshot5_stream}; | |
| CheckStreamScanResult(table_scan, table_read, expected_snapshot_ids, expected_arrays); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithPKWithMorStreamFromSnapshot2) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_mor.db/pk_table_scan_and_read_mor/"; | |
| // use from-snapshot mode, first plan is empty, snapshot 2 in second plan only includes delta | |
| // files | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "2") | |
| .AddOption(Options::SCAN_MODE, "from-snapshot") | |
| .WithStreamingMode(true); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| std::vector<std::optional<int64_t>> expected_snapshot_ids = {std::nullopt, 2, 4, 5}; | |
| auto expected_snapshot2_stream = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [2, "Alice", 10, 1, 19.1], | |
| [3, "Tony", 10, 0, 14.1] | |
| ])") | |
| .ValueOrDie()); | |
| auto expected_snapshot4_stream = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [2, "Alex", 10, 0, 21.2], | |
| [0, "Marco", 10, 0, 21.1], | |
| [0, "Skye", 10, 0, 21.0] | |
| ])") | |
| .ValueOrDie()); | |
| auto expected_snapshot5_stream = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [3, "Alex", 10, 0, 31.2], | |
| [0, "Marco2", 10, 0, 31.1], | |
| [0, "Skye2", 10, 0, 31.0] | |
| ])") | |
| .ValueOrDie()); | |
| std::vector<std::shared_ptr<arrow::ChunkedArray>> expected_arrays = { | |
| nullptr, expected_snapshot2_stream, expected_snapshot4_stream, expected_snapshot5_stream}; | |
| CheckStreamScanResult(table_scan, table_read, expected_snapshot_ids, expected_arrays); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithPKWithMorStreamFromSnapshot5WithPredicate) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_mor.db/pk_table_scan_and_read_mor/"; | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "5") | |
| .AddOption(Options::SCAN_MODE, "from-snapshot") | |
| .WithStreamingMode(true); | |
| auto predicate = PredicateBuilder::GreaterThan(/*field_index=*/3, /*field_name=*/"f3", | |
| FieldType::DOUBLE, Literal(50.0)); | |
| scan_context_builder.SetPredicate(predicate); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| read_context_builder.SetPredicate(predicate); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| // first plan is snapshot 5 with empty plan, second plan is snapshot 5 with delta files | |
| // (predicates take no effects in delta mode) | |
| std::vector<std::optional<int64_t>> expected_snapshot_ids = {std::nullopt, 5}; | |
| // for streaming data split, return array with row_kind | |
| auto expected_snapshot5_stream = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [3, "Alex", 10, 0, 31.2], | |
| [0, "Marco2", 10, 0, 31.1], | |
| [0, "Skye2", 10, 0, 31.0] | |
| ])") | |
| .ValueOrDie()); | |
| std::vector<std::shared_ptr<arrow::ChunkedArray>> expected_arrays = {nullptr, | |
| expected_snapshot5_stream}; | |
| CheckStreamScanResult(table_scan, table_read, expected_snapshot_ids, expected_arrays); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithPKWithMorStreamPredicateFilter) { | |
| std::string table_path = paimon::test::GetDataDir() + FileFormat() + | |
| "/pk_table_scan_and_read_mor.db/pk_table_scan_and_read_mor/"; | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "5") | |
| .AddOption(Options::SCAN_MODE, "from-snapshot") | |
| .WithStreamingMode(true); | |
| std::shared_ptr<Predicate> predicate = PredicateBuilder::GreaterThan( | |
| /*field_index=*/3, /*field_name=*/"f3", FieldType::DOUBLE, Literal(31.05)); | |
| scan_context_builder.SetPredicate(predicate); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ScanContext> scan_context, | |
| FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableScan> table_scan, | |
| TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| read_context_builder.SetPredicate(predicate).EnablePredicateFilter(true); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableRead> table_read, | |
| TableRead::Create(std::move(read_context))); | |
| std::vector<std::optional<int64_t>> expected_snapshot_ids = {std::nullopt, 5}; | |
| auto expected_snapshot5_stream = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [3, "Alex", 10, 0, 31.2], | |
| [0, "Marco2", 10, 0, 31.1] | |
| ])") | |
| .ValueOrDie()); | |
| std::vector<std::shared_ptr<arrow::ChunkedArray>> expected_arrays = {nullptr, | |
| expected_snapshot5_stream}; | |
| CheckStreamScanResult(table_scan, table_read, expected_snapshot_ids, expected_arrays); | |
| } | |
| // test first row merge engine | |
| TEST_P(ScanAndReadInteTest, TestWithPKWithFirstRowBatchScanSnapshot5) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = | |
| paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_first_row.db/pk_table_scan_and_read_first_row/"; | |
| // normal batch scan case for pk+first row, all data in level 0 is filtered out | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "5"); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(auto read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 5); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(result_plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Alice", 10, 1, 11.1], | |
| [0, "Two roads diverged in a wood, and I took the one less traveled by, And that has made all the difference.", 10, 1, 11.0], | |
| [0, "Alex", 10, 0, 16.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "David", 10, 0, 17.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Tony", 10, 0, 14.1], | |
| [0, "Lucy", 20, 1, 14.1], | |
| [0, "Paul", 20, 1, 18.1] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithPKWithFirstRowStreamFromSnapshot3) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = | |
| paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_first_row.db/pk_table_scan_and_read_first_row/"; | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "3") | |
| .AddOption(Options::SCAN_MODE, "from-snapshot-full") | |
| .WithStreamingMode(true); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| // from a compact snapshot | |
| std::vector<std::optional<int64_t>> expected_snapshot_ids = {3, 4, 5}; | |
| auto expected_snapshot3_batch = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Alice", 10, 1, 11.1], | |
| [0, "Two roads diverged in a wood, and I took the one less traveled by, And that has made all the difference.", 10, 1, 11.0], | |
| [0, "Alex", 10, 0, 16.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "David", 10, 0, 17.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Tony", 10, 0, 14.1], | |
| [0, "Lucy", 20, 1, 14.1], | |
| [0, "Paul", 20, 1, 18.1] | |
| ])") | |
| .ValueOrDie()); | |
| auto expected_snapshot4_stream = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [2, "Alex", 10, 0, 21.2], | |
| [0, "Marco", 10, 0, 21.1], | |
| [0, "Skye", 10, 0, 21.0] | |
| ])") | |
| .ValueOrDie()); | |
| // delete record is ignored | |
| auto expected_snapshot5_stream = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Marco2", 10, 0, 31.1], | |
| [0, "Skye2", 10, 0, 31.0] | |
| ])") | |
| .ValueOrDie()); | |
| std::vector<std::shared_ptr<arrow::ChunkedArray>> expected_arrays = { | |
| expected_snapshot3_batch, expected_snapshot4_stream, expected_snapshot5_stream}; | |
| CheckStreamScanResult(table_scan, table_read, expected_snapshot_ids, expected_arrays); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithPKWithFirstRowStreamFromSnapshot5) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = | |
| paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_first_row.db/pk_table_scan_and_read_first_row/"; | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_MODE, "latest-full").WithStreamingMode(true); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| // from an append snapshot, first plan is snapshot5 with merge, all level 0 data is merged | |
| std::vector<std::optional<int64_t>> expected_snapshot_ids = {5}; | |
| auto expected_snapshot5_batch = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Alice", 10, 1, 11.1], | |
| [0, "Two roads diverged in a wood, and I took the one less traveled by, And that has made all the difference.", 10, 1, 11.0], | |
| [0, "Alex", 10, 0, 16.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "David", 10, 0, 17.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Marco", 10, 0, 21.1], | |
| [0, "Marco2", 10, 0, 31.1], | |
| [0, "Skye", 10, 0, 21.0], | |
| [0, "Skye2", 10, 0, 31.0], | |
| [0, "Tony", 10, 0, 14.1], | |
| [0, "Lucy", 20, 1, 14.1], | |
| [0, "Paul", 20, 1, 18.1] | |
| ])") | |
| .ValueOrDie()); | |
| std::vector<std::shared_ptr<arrow::ChunkedArray>> expected_arrays = {expected_snapshot5_batch}; | |
| CheckStreamScanResult(table_scan, table_read, expected_snapshot_ids, expected_arrays); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithPKWith09VersionDvBatchScanLatestSnapshot) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + "/pk_09.db/pk_09/"; | |
| // normal batch scan case for pk+dv (09 version) | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.SetLimit(2); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(auto read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 8); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(result_plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Alex", 10, 0, 16.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "David", 10, 0, 17.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Alice", 10, 1, 21.1], | |
| [0, "Two roads diverged in a wood, and I took the one less traveled by, And that has made all the difference.", 10, 1, 11.0], | |
| [0, "Whether I shall turn out to be the hero of my own life.", 10, 1, 19.1], | |
| [0, "Lucy", 20, 1, 14.1], | |
| [0, "Paul", 20, 1, 18.1] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithEmptyPartitionValue) { | |
| auto file_format = FileFormat(); | |
| auto check_result = | |
| [&](const std::string& table_path, | |
| const std::vector<std::map<std::string, std::string>>& partition_filters, | |
| const std::shared_ptr<arrow::ChunkedArray>& expected) { | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.SetPartitionFilter(partition_filters); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(auto read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 1); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, | |
| table_read->CreateReader(result_plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| }; | |
| auto expected_without_partition_filter = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "2025", 500, -1, 15.1], | |
| [0, "2024", 100, -2, 11.1], | |
| [0, "2024", 300, 1, 13.1], | |
| [0, null, 200, -3, 12.1], | |
| [0, null, 50, 1, 14.1], | |
| [0, null, 100, 2, 16.1] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected_without_partition_filter); | |
| { | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/append_with_empty_partition.db/append_with_empty_partition/"; | |
| check_result(table_path, {}, expected_without_partition_filter); | |
| check_result(table_path, {{{"f0", "2025"}}}, | |
| expected_without_partition_filter->Slice(0, 1)); | |
| check_result(table_path, {{{"f0", "__DEFAULT_PARTITION__"}}}, | |
| expected_without_partition_filter->Slice(3, 3)); | |
| check_result(table_path, {{{"f0", "__DEFAULT_PARTITION__"}}, {{"f0", "2024"}}}, | |
| expected_without_partition_filter->Slice(1, 5)); | |
| } | |
| { | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/append_with_empty_partition_with_specific_name.db/" | |
| "append_with_empty_partition_with_specific_name/"; | |
| // check with specific partition name | |
| check_result(table_path, {}, expected_without_partition_filter); | |
| check_result(table_path, {{{"f0", "2025"}}}, | |
| expected_without_partition_filter->Slice(0, 1)); | |
| check_result(table_path, {{{"f0", "__HIVE_DEFAULT_PARTITION__"}}}, | |
| expected_without_partition_filter->Slice(3, 3)); | |
| check_result(table_path, {{{"f0", "__HIVE_DEFAULT_PARTITION__"}}, {{"f0", "2024"}}}, | |
| expected_without_partition_filter->Slice(1, 5)); | |
| } | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithMultipleEmptyPartitionValue) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/append_with_empty_partition_with_empty_value.db/" | |
| "append_with_empty_partition_with_empty_value/"; | |
| auto check_result = | |
| [&](const std::vector<std::map<std::string, std::string>>& partition_filters, | |
| const std::shared_ptr<arrow::ChunkedArray>& expected) { | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.SetPartitionFilter(partition_filters); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(auto read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 1); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, | |
| table_read->CreateReader(result_plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| }; | |
| auto expected_without_partition_filter = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "", 200, -3, 12.1], | |
| [0, "", 50, 1, 14.1], | |
| [0, "2025", 500, -1, 15.1], | |
| [0, " ", 100, 0, 15.1], | |
| [0, "2024", 100, -2, 11.1], | |
| [0, "2024", 300, 1, 13.1], | |
| [0, null, 100, 2, 16.1] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected_without_partition_filter); | |
| check_result({}, expected_without_partition_filter); | |
| check_result({{{"f0", "2025"}}}, expected_without_partition_filter->Slice(2, 1)); | |
| check_result({{{"f0", "__DEFAULT_PARTITION__"}}}, | |
| expected_without_partition_filter->Slice(6, 1)); | |
| check_result({{{"f0", ""}}}, expected_without_partition_filter->Slice(0, 2)); | |
| check_result({{{"f0", " "}}}, expected_without_partition_filter->Slice(3, 1)); | |
| check_result({{{"f0", ""}}, {{"f0", "2025"}}}, expected_without_partition_filter->Slice(0, 3)); | |
| check_result({{{"f0", "2024"}}, {{"f0", "__DEFAULT_PARTITION__"}}}, | |
| expected_without_partition_filter->Slice(4, 3)); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestMemoryUse) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + "/append_09.db/append_09/"; | |
| // scan | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "1"); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 1); | |
| auto splits = result_plan->Splits(); | |
| ASSERT_EQ(3, splits.size()); | |
| // read | |
| std::shared_ptr<MemoryPool> read_pool = GetMemoryPool(); | |
| { | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| read_context_builder.WithMemoryPool(read_pool); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, | |
| read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(splits)); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Alice", 10, 1, 11.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Tony", 10, 0, 14.1], | |
| [0, "Lucy", 20, 1, 14.1] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| } | |
| // check all memory is released | |
| ASSERT_GT(read_pool->MaxMemoryUsage(), 0); | |
| ASSERT_EQ(read_pool->CurrentUsage(), 0); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestPkScanWithPostponeBucket) { | |
| auto file_format = FileFormat(); | |
| auto test_dir = UniqueTestDirectory::Create("local"); | |
| arrow::FieldVector fields = {arrow::field("f0", arrow::utf8()), | |
| arrow::field("f1", arrow::int32()), | |
| arrow::field("f2", arrow::float64())}; | |
| auto field_with_row_kind = fields; | |
| field_with_row_kind.insert(field_with_row_kind.begin(), | |
| arrow::field("_VALUE_KIND", arrow::int8())); | |
| auto schema = arrow::schema(fields); | |
| std::map<std::string, std::string> options = {{Options::FILE_FORMAT, file_format}, | |
| {Options::TARGET_FILE_SIZE, "1024"}, | |
| {Options::BUCKET, "-2"}, | |
| {Options::FILE_SYSTEM, "local"}}; | |
| ASSERT_OK_AND_ASSIGN(auto helper, | |
| TestHelper::Create(test_dir->Str(), schema, /*partition_keys=*/{"f1"}, | |
| /*primary_keys=*/{"f0", "f1"}, options, | |
| /*is_streaming_mode=*/true)); | |
| std::string table_path = test_dir->Str() + "/foo.db/bar"; | |
| int64_t commit_identifier = 0; | |
| // write batch1 | |
| std::string data1 = R"([ | |
| ["banana", 1, 3.5], | |
| ["dog", 1, 2000.5], | |
| ["lucy", 1, 10000.5], | |
| ["mouse", 1, 10.5] | |
| ])"; | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch1, | |
| TestHelper::MakeRecordBatch( | |
| arrow::struct_(fields), data1, | |
| /*partition_map=*/std::map<std::string, std::string>({{"f1", "1"}}), | |
| /*bucket=*/-2, {})); | |
| ASSERT_OK_AND_ASSIGN(auto commit_msgs1, | |
| helper->WriteAndCommit(std::move(batch1), commit_identifier++, | |
| /*expected_commit_messages=*/std::nullopt)); | |
| // write batch2 | |
| std::string data2 = R"([ | |
| ["Paul", 2, 12.1], | |
| ["Cathy", 2, 13.1], | |
| ["Emily", 2, 14.1], | |
| ["Cathy", 2, 13.1] | |
| ])"; | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch2, | |
| TestHelper::MakeRecordBatch( | |
| arrow::struct_(fields), data2, | |
| /*partition_map=*/std::map<std::string, std::string>({{"f1", "2"}}), | |
| /*bucket=*/-2, | |
| {RecordBatch::RowKind::INSERT, RecordBatch::RowKind::INSERT, | |
| RecordBatch::RowKind::INSERT, RecordBatch::RowKind::DELETE})); | |
| ASSERT_OK_AND_ASSIGN(auto commit_msgs2, | |
| helper->WriteAndCommit(std::move(batch2), commit_identifier++, | |
| /*expected_commit_messages=*/std::nullopt)); | |
| std::vector<std::string> subdirs = {"f1=1/bucket-postpone", "f1=2/bucket-postpone"}; | |
| CheckPostponeFile(table_path, subdirs); | |
| { | |
| // batch scan | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.WithStreamingMode(false); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 2); | |
| ASSERT_TRUE(result_plan->Splits().empty()); | |
| } | |
| { | |
| // stream scan: from snapshot 1 | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "1").WithStreamingMode(true); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, | |
| read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| std::vector<std::optional<int64_t>> expected_snapshot_ids = {std::nullopt, 1, 2}; | |
| auto expected_snapshot1 = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(field_with_row_kind), R"([ | |
| [0, "banana", 1, 3.5], | |
| [0, "dog", 1, 2000.5], | |
| [0, "lucy", 1, 10000.5], | |
| [0, "mouse", 1, 10.5] | |
| ])") | |
| .ValueOrDie()); | |
| auto expected_snapshot2 = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(field_with_row_kind), R"([ | |
| [0, "Paul", 2, 12.1], | |
| [0, "Cathy", 2, 13.1], | |
| [0, "Emily", 2, 14.1], | |
| [3, "Cathy", 2, 13.1] | |
| ])") | |
| .ValueOrDie()); | |
| std::vector<std::shared_ptr<arrow::ChunkedArray>> expected_arrays = { | |
| nullptr, expected_snapshot1, expected_snapshot2}; | |
| CheckStreamScanResult(table_scan, table_read, expected_snapshot_ids, expected_arrays); | |
| } | |
| { | |
| // stream scan: from snapshot 2 | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "2").WithStreamingMode(true); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, | |
| read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| std::vector<std::optional<int64_t>> expected_snapshot_ids = {std::nullopt, 2}; | |
| auto expected_snapshot2 = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(field_with_row_kind), R"([ | |
| [0, "Paul", 2, 12.1], | |
| [0, "Cathy", 2, 13.1], | |
| [0, "Emily", 2, 14.1], | |
| [3, "Cathy", 2, 13.1] | |
| ])") | |
| .ValueOrDie()); | |
| std::vector<std::shared_ptr<arrow::ChunkedArray>> expected_arrays = {nullptr, | |
| expected_snapshot2}; | |
| CheckStreamScanResult(table_scan, table_read, expected_snapshot_ids, expected_arrays); | |
| } | |
| { | |
| // stream scan: from snapshot full | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "1") | |
| .AddOption(Options::SCAN_MODE, "from-snapshot-full") | |
| .WithStreamingMode(true); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, | |
| read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| std::vector<std::optional<int64_t>> expected_snapshot_ids = {1, 2}; | |
| auto expected_snapshot1 = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(field_with_row_kind), R"([ | |
| [0, "banana", 1, 3.5], | |
| [0, "dog", 1, 2000.5], | |
| [0, "lucy", 1, 10000.5], | |
| [0, "mouse", 1, 10.5] | |
| ])") | |
| .ValueOrDie()); | |
| auto expected_snapshot2 = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(field_with_row_kind), R"([ | |
| [0, "Paul", 2, 12.1], | |
| [0, "Cathy", 2, 13.1], | |
| [0, "Emily", 2, 14.1], | |
| [3, "Cathy", 2, 13.1] | |
| ])") | |
| .ValueOrDie()); | |
| std::vector<std::shared_ptr<arrow::ChunkedArray>> expected_arrays = {expected_snapshot1, | |
| expected_snapshot2}; | |
| CheckStreamScanResult(table_scan, table_read, expected_snapshot_ids, expected_arrays); | |
| } | |
| } | |
| TEST_P(ScanAndReadInteTest, TestScanWithPredicateAndReadWithUnorderedFieldForParquet) { | |
| auto file_format = FileFormat(); | |
| if (file_format != "parquet") { | |
| return; | |
| } | |
| std::string table_path = | |
| paimon::test::GetDataDir() + "parquet/parquet_append_table.db/parquet_append_table"; | |
| ScanContextBuilder scan_context_builder(table_path); | |
| auto predicate = PredicateBuilder::LessThan( | |
| /*field_index=*/3, /*field_name=*/"f4", FieldType::INT, Literal(300006)); | |
| scan_context_builder.SetPredicate(predicate); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 2); | |
| ASSERT_EQ(result_plan->Splits().size(), 1); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| read_context_builder.SetReadFieldNames({"f10", "f8", "f4", "f13"}); | |
| ASSERT_OK_AND_ASSIGN(auto read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(result_plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON( | |
| arrow::struct_( | |
| {arrow::field("_VALUE_KIND", arrow::int8()), | |
| arrow::field("f10", | |
| arrow::map(arrow::list(arrow::float32()), | |
| arrow::struct_({arrow::field("f0", arrow::boolean()), | |
| arrow::field("f1", arrow::int64())}))), | |
| arrow::field("f8", arrow::utf8()), arrow::field("f4", arrow::int32()), | |
| arrow::field("f13", arrow::decimal128(2, 2))}), | |
| R"([ | |
| [0, [[[5.11, 5.21], [true, 61]]], "s31", 300001, "0.91"], | |
| [0, [[[5.12, 5.22], [false, 62]]], "s32", 300002, "0.92"], | |
| [0, null, "s33", 300003, "0.93"], | |
| [0, [[[5.141, 5.241], [false, 641]], [[5.14, 5.24], [false, 64]]], "s34", 300004, "0.94"], | |
| [0, [[[5.15, 5.25], [true, 65]]], "s35", 300005, "0.95"], | |
| [0, [[[5.16, 5.26], null]], "s36", 300006, "0.96"] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestPkSchemaEvolutionScanWithRenamedPkPredicate) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_table_with_alter_table.db/pk_table_with_alter_table/"; | |
| auto predicate = PredicateBuilder::GreaterThan( | |
| /*field_index=*/2, /*field_name=*/"key_2", FieldType::INT, Literal(500)); | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "6"); | |
| scan_context_builder.SetPredicate(predicate); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 6); | |
| ASSERT_EQ(result_plan->Splits().size(), 1); | |
| size_t data_file_count = 0; | |
| for (const auto& split : result_plan->Splits()) { | |
| auto data_split = std::dynamic_pointer_cast<DataSplitImpl>(split); | |
| ASSERT_TRUE(data_split); | |
| data_file_count += data_split->data_files_.size(); | |
| } | |
| ASSERT_EQ(data_file_count, 1); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| read_context_builder.SetReadFieldNames({"key1", "k", "key_2", "c", "d", "a", "key0", "e"}); | |
| ASSERT_OK_AND_ASSIGN(auto read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(result_plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON( | |
| arrow::struct_({arrow::field("_VALUE_KIND", arrow::int8()), | |
| arrow::field("key1", arrow::int32()), arrow::field("k", arrow::utf8()), | |
| arrow::field("key_2", arrow::int32()), | |
| arrow::field("c", arrow::int32()), arrow::field("d", arrow::int32()), | |
| arrow::field("a", arrow::int32()), arrow::field("key0", arrow::int32()), | |
| arrow::field("e", arrow::int32())}), | |
| R"([ | |
| [0, 1, "Two roads diverged in a wood, and I took the one less traveled by, And that has made all the difference.", 2, 4, null, 6, 0, null], | |
| [0, 1, "Alice", 12, 94, null, 96, 0, null], | |
| [0, 1, "Paul", 502, 504, 508, 506, 0, 509] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestAppendTableWithMultipleFileFormat) { | |
| auto file_format = FileFormat(); | |
| if (file_format != "parquet") { | |
| return; | |
| } | |
| std::string table_path = | |
| paimon::test::GetDataDir() + | |
| "/append_table_with_multiple_file_format.db/append_table_with_multiple_file_format"; | |
| // scan | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "2"); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 2); | |
| auto splits = result_plan->Splits(); | |
| ASSERT_EQ(1, splits.size()); | |
| // read | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(splits)); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Alice", 10, 1, 11.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Tony", 10, 0, 14.1], | |
| [0, "Lucy", 20, 1, 14.1], | |
| [0, "Skye", 20, 1, 19.1], | |
| [0, "Bob", 10, 0, 20.1] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestPkDvTableIndexInDataAndNoExternalPath) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_dv_index_in_data_no_external.db/pk_dv_index_in_data_no_external"; | |
| // scan | |
| ScanContextBuilder scan_context_builder(table_path); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 4); | |
| auto splits = result_plan->Splits(); | |
| ASSERT_EQ(3, splits.size()); | |
| // read | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(splits)); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Two roads diverged in a wood, and I took the one less traveled by, And that has made all the difference.", 10, 1, 11.0], | |
| [0, "Alice", 10, 1, 19.1], | |
| [0, "Alex", 10, 0, 16.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "David", 10, 0, 17.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Lucy", 20, 1, 14.1], | |
| [0, "Paul", 20, 1, 18.1] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestPkDvTableIndexNotInDataAndNoExternalPath) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = | |
| paimon::test::GetDataDir() + file_format + | |
| "/pk_dv_index_not_in_data_no_external.db/pk_dv_index_not_in_data_no_external"; | |
| // scan | |
| ScanContextBuilder scan_context_builder(table_path); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 4); | |
| auto splits = result_plan->Splits(); | |
| ASSERT_EQ(3, splits.size()); | |
| // read | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(splits)); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Two roads diverged in a wood, and I took the one less traveled by, And that has made all the difference.", 10, 1, 11.0], | |
| [0, "Alice", 10, 1, 19.1], | |
| [0, "Alex", 10, 0, 16.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "David", 10, 0, 17.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Lucy", 20, 1, 14.1], | |
| [0, "Paul", 20, 1, 18.1] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestPkDvTableIndexNotInDataAndWithExternalPath) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = | |
| paimon::test::GetDataDir() + file_format + | |
| "/pk_dv_index_not_in_data_with_external.db/pk_dv_index_not_in_data_with_external"; | |
| std::string external_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_dv_index_not_in_data_with_external.db/external"; | |
| // scan | |
| ScanContextBuilder scan_context_builder(table_path); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 4); | |
| auto splits = result_plan->Splits(); | |
| ASSERT_EQ(3, splits.size()); | |
| AdjustSplitWithExternalPath("FILE:/tmp/external", external_path, /*adjust_index=*/false, | |
| &splits); | |
| // read | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(splits)); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Two roads diverged in a wood, and I took the one less traveled by, And that has made all the difference.", 10, 1, 11.0], | |
| [0, "Alice", 10, 1, 19.1], | |
| [0, "Alex", 10, 0, 16.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "David", 10, 0, 17.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Lucy", 20, 1, 14.1], | |
| [0, "Paul", 20, 1, 18.1] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestScanAndReadWithDisableIndex) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = | |
| paimon::test::GetDataDir() + file_format + "/append_with_bitmap.db/append_with_bitmap"; | |
| auto predicate = | |
| PredicateBuilder::Equal(/*field_index=*/0, /*field_name=*/"f0", FieldType::STRING, | |
| Literal(FieldType::STRING, "Alice0", 6)); | |
| // scan | |
| ScanContextBuilder scan_context_builder(table_path); | |
| ASSERT_OK_AND_ASSIGN( | |
| auto scan_context, | |
| scan_context_builder.AddOption("file-index.read.enabled", "false").Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 1); | |
| auto splits = result_plan->Splits(); | |
| ASSERT_EQ(1, splits.size()); | |
| // read | |
| ReadContextBuilder read_context_builder(table_path); | |
| read_context_builder.AddOption("file-index.read.enabled", "false"); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(splits)); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result, when file-index.read.enabled = false, index will be ignored | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Alice", 10, 1, 11.1], | |
| [0, "Bob", 10, 1, 12.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Tony", 10, 0, 14.1], | |
| [0, "Lucy", 20, 1, 15.1], | |
| [0, "Bob", 10, 1, 16.1], | |
| [0, "Tony", 20, 0, 17.1], | |
| [0, "Alice", 20, null, 18.1] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| } | |
| TEST(ScanAndReadFileIndexInteTest, TestPkDvEmbeddedBitmapFiltersScanAndRead) { | |
| std::unique_ptr<UniqueTestDirectory> dir = UniqueTestDirectory::Create("local"); | |
| ASSERT_NE(nullptr, dir); | |
| const std::string table_path = PathUtil::JoinPath(dir->Str(), "foo.db/bar"); | |
| const arrow::FieldVector fields = {arrow::field("id", arrow::int32()), | |
| arrow::field("indexed_value", arrow::int32()), | |
| arrow::field("payload", arrow::utf8())}; | |
| const std::shared_ptr<arrow::Schema> schema = arrow::schema(fields); | |
| const std::map<std::string, std::string> options = { | |
| {Options::FILE_FORMAT, "parquet"}, | |
| {Options::FILE_SYSTEM, "local"}, | |
| {Options::BUCKET, "1"}, | |
| {Options::BUCKET_KEY, "id"}, | |
| {Options::DELETION_VECTORS_ENABLED, "true"}, | |
| {Options::COMMIT_FORCE_COMPACT, "true"}, | |
| {Options::FILE_INDEX_READ_ENABLED, "true"}, | |
| {Options::FILE_INDEX_IN_MANIFEST_THRESHOLD, "1MB"}, | |
| {"file-index.bitmap.columns", "indexed_value"}, | |
| }; | |
| ASSERT_OK_AND_ASSIGN( | |
| std::unique_ptr<TestHelper> helper, | |
| TestHelper::Create(dir->Str(), schema, /*partition_keys=*/{}, /*primary_keys=*/{"id"}, | |
| options, /*is_streaming_mode=*/true, /*ignore_if_exists=*/false, | |
| PathUtil::JoinPath(dir->Str(), "tmp"))); | |
| ASSERT_OK_AND_ASSIGN( | |
| std::unique_ptr<RecordBatch> batch, | |
| TestHelper::MakeRecordBatch(arrow::struct_(fields), | |
| R"([[1, 10, "one"], [2, 20, "two"], [3, 30, "three"]])", | |
| /*partition_map=*/{}, /*bucket=*/0, /*row_kinds=*/{})); | |
| ASSERT_OK_AND_ASSIGN( | |
| std::vector<std::shared_ptr<CommitMessage>> commit_messages, | |
| helper->WriteAndCommit(std::move(batch), /*commit_identifier=*/0, std::nullopt)); | |
| ASSERT_FALSE(commit_messages.empty()); | |
| // 15 falls inside the file's [10, 30] min/max range, so statistics alone must retain the | |
| // file. Disabling file-index reads verifies that baseline explicitly. | |
| const std::shared_ptr<Predicate> absent_predicate = PredicateBuilder::Equal( | |
| /*field_index=*/1, /*field_name=*/"indexed_value", FieldType::INT, Literal(15)); | |
| ScanContextBuilder stats_only_scan_builder(table_path); | |
| stats_only_scan_builder.SetOptions(options) | |
| .AddOption(Options::FILE_INDEX_READ_ENABLED, "false") | |
| .SetPredicate(absent_predicate); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ScanContext> stats_only_scan_context, | |
| stats_only_scan_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableScan> stats_only_scan, | |
| TableScan::Create(std::move(stats_only_scan_context))); | |
| ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> stats_only_plan, stats_only_scan->CreatePlan()); | |
| ASSERT_FALSE(stats_only_plan->Splits().empty()); | |
| // The embedded bitmap knows that 15 is absent and eliminates the file during scan planning. | |
| ScanContextBuilder indexed_scan_builder(table_path); | |
| indexed_scan_builder.SetOptions(options).SetPredicate(absent_predicate); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ScanContext> indexed_scan_context, | |
| indexed_scan_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableScan> indexed_scan, | |
| TableScan::Create(std::move(indexed_scan_context))); | |
| ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> indexed_plan, indexed_scan->CreatePlan()); | |
| ASSERT_TRUE(indexed_plan->Splits().empty()); | |
| // For a present value, scan retains the embedded-index file and read applies its precise | |
| // bitmap. Residual predicate filtering stays disabled so only file-index selection can remove | |
| // the other rows. | |
| const std::shared_ptr<Predicate> present_predicate = PredicateBuilder::Equal( | |
| /*field_index=*/1, /*field_name=*/"indexed_value", FieldType::INT, Literal(20)); | |
| ScanContextBuilder retained_scan_builder(table_path); | |
| retained_scan_builder.SetOptions(options).SetPredicate(present_predicate); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ScanContext> retained_scan_context, | |
| retained_scan_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableScan> retained_scan, | |
| TableScan::Create(std::move(retained_scan_context))); | |
| ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> retained_plan, retained_scan->CreatePlan()); | |
| ASSERT_FALSE(retained_plan->Splits().empty()); | |
| ReadContextBuilder read_context_builder(table_path); | |
| read_context_builder.SetOptions(options) | |
| .SetPredicate(present_predicate) | |
| .EnablePredicateFilter(false); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableRead> table_read, | |
| TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<BatchReader> batch_reader, | |
| table_read->CreateReader(retained_plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> actual, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| const std::shared_ptr<arrow::DataType> result_type = arrow::struct_( | |
| {arrow::field("_VALUE_KIND", arrow::int8()), arrow::field("id", arrow::int32()), | |
| arrow::field("indexed_value", arrow::int32()), arrow::field("payload", arrow::utf8())}); | |
| const std::shared_ptr<arrow::ChunkedArray> expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(result_type, R"([[0, 2, 20, "two"]])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected->Equals(actual)) << actual->ToString(); | |
| } | |
| TEST(ScanAndReadFileIndexInteTest, TestPkDvReconstructsFileIndexPredicateAfterSchemaEvolution) { | |
| std::unique_ptr<UniqueTestDirectory> dir = UniqueTestDirectory::Create("local"); | |
| ASSERT_NE(nullptr, dir); | |
| const std::string table_path = PathUtil::JoinPath(dir->Str(), "foo.db/bar"); | |
| const arrow::FieldVector fields = {arrow::field("id", arrow::int32()), | |
| arrow::field("indexed_value", arrow::int32()), | |
| arrow::field("payload", arrow::utf8())}; | |
| const std::shared_ptr<arrow::Schema> schema = arrow::schema(fields); | |
| std::map<std::string, std::string> options = { | |
| {Options::FILE_FORMAT, "parquet"}, | |
| {Options::FILE_SYSTEM, "local"}, | |
| {Options::BUCKET, "1"}, | |
| {Options::BUCKET_KEY, "id"}, | |
| {Options::DELETION_VECTORS_ENABLED, "true"}, | |
| {Options::COMMIT_FORCE_COMPACT, "true"}, | |
| {Options::FILE_INDEX_READ_ENABLED, "true"}, | |
| {Options::FILE_INDEX_IN_MANIFEST_THRESHOLD, "1MB"}, | |
| {"file-index.bitmap.columns", "indexed_value"}, | |
| }; | |
| ASSERT_OK_AND_ASSIGN( | |
| std::unique_ptr<TestHelper> helper, | |
| TestHelper::Create(dir->Str(), schema, /*partition_keys=*/{}, /*primary_keys=*/{"id"}, | |
| options, /*is_streaming_mode=*/true, /*ignore_if_exists=*/false, | |
| PathUtil::JoinPath(dir->Str(), "tmp"))); | |
| ASSERT_OK_AND_ASSIGN( | |
| std::unique_ptr<RecordBatch> batch, | |
| TestHelper::MakeRecordBatch(arrow::struct_(fields), | |
| R"([[1, 10, "one"], [2, 20, "two"], [3, 30, "three"]])", | |
| /*partition_map=*/{}, /*bucket=*/0, /*row_kinds=*/{})); | |
| ASSERT_OK(helper->WriteAndCommit(std::move(batch), /*commit_identifier=*/0, std::nullopt)); | |
| helper.reset(); | |
| // Keep field IDs and types unchanged while renaming the indexed value field. The old file and | |
| // its embedded bitmap still use "indexed_value", whereas scans use "renamed_value". | |
| options["file-index.bitmap.columns"] = "renamed_value"; | |
| const std::vector<DataField> evolved_fields = { | |
| DataField(0, arrow::field("id", arrow::int32(), /*nullable=*/false)), | |
| DataField(1, arrow::field("renamed_value", arrow::int32())), | |
| DataField(2, arrow::field("payload", arrow::utf8()))}; | |
| ASSERT_OK(TestHelper::WriteNextSchema(dir->GetFileSystem(), table_path, evolved_fields, | |
| /*highest_field_id=*/2, options)); | |
| const std::shared_ptr<Predicate> predicate = PredicateBuilder::Equal( | |
| /*field_index=*/1, /*field_name=*/"renamed_value", FieldType::INT, Literal(15)); | |
| // The evolved min/max range is still [10, 30], so statistics alone retain the old file. | |
| ScanContextBuilder stats_only_scan_builder(table_path); | |
| stats_only_scan_builder.SetOptions(options) | |
| .AddOption(Options::FILE_INDEX_READ_ENABLED, "false") | |
| .SetPredicate(predicate); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ScanContext> stats_only_scan_context, | |
| stats_only_scan_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableScan> stats_only_scan, | |
| TableScan::Create(std::move(stats_only_scan_context))); | |
| ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> stats_only_plan, stats_only_scan->CreatePlan()); | |
| ASSERT_FALSE(stats_only_plan->Splits().empty()); | |
| // Reconstructing the predicate to schema-0 lets its embedded bitmap prove that 15 is absent. | |
| ScanContextBuilder indexed_scan_builder(table_path); | |
| indexed_scan_builder.SetOptions(options).SetPredicate(predicate); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ScanContext> indexed_scan_context, | |
| indexed_scan_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableScan> indexed_scan, | |
| TableScan::Create(std::move(indexed_scan_context))); | |
| ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> indexed_plan, indexed_scan->CreatePlan()); | |
| ASSERT_TRUE(indexed_plan->Splits().empty()); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestPkDvTableIndexInDataAndWithExternalPath) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = | |
| paimon::test::GetDataDir() + file_format + | |
| "/pk_dv_index_in_data_with_external.db/pk_dv_index_in_data_with_external"; | |
| std::string external_path = | |
| paimon::test::GetDataDir() + file_format + "/pk_dv_index_in_data_with_external.db/external"; | |
| // scan | |
| ScanContextBuilder scan_context_builder(table_path); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 4); | |
| auto splits = result_plan->Splits(); | |
| ASSERT_EQ(3, splits.size()); | |
| AdjustSplitWithExternalPath("FILE:/tmp/external", external_path, /*adjust_index=*/true, | |
| &splits); | |
| // read | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(splits)); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Two roads diverged in a wood, and I took the one less traveled by, And that has made all the difference.", 10, 1, 11.0], | |
| [0, "Alice", 10, 1, 19.1], | |
| [0, "Alex", 10, 0, 16.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "David", 10, 0, 17.1], | |
| [0, "Emily", 10, 0, 13.1], | |
| [0, "Lucy", 20, 1, 14.1], | |
| [0, "Paul", 20, 1, 18.1] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestTimestampType) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/append_with_multiple_ts_precision_and_timezone.db" | |
| "/append_with_multiple_ts_precision_and_timezone/"; | |
| ScanContextBuilder scan_context_builder(table_path); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 1); | |
| auto splits = result_plan->Splits(); | |
| ASSERT_EQ(1, splits.size()); | |
| // read | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(splits)); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result | |
| auto timezone = DateTimeUtils::GetLocalTimezoneName(); | |
| arrow::FieldVector fields = { | |
| arrow::field("_VALUE_KIND", arrow::int8()), | |
| arrow::field("ts_sec", arrow::timestamp(arrow::TimeUnit::SECOND)), | |
| arrow::field("ts_milli", arrow::timestamp(arrow::TimeUnit::MILLI)), | |
| arrow::field("ts_micro", arrow::timestamp(arrow::TimeUnit::MICRO)), | |
| arrow::field("ts_nano", arrow::timestamp(arrow::TimeUnit::NANO)), | |
| arrow::field("ts_tz_sec", arrow::timestamp(arrow::TimeUnit::SECOND, timezone)), | |
| arrow::field("ts_tz_milli", arrow::timestamp(arrow::TimeUnit::MILLI, timezone)), | |
| arrow::field("ts_tz_micro", arrow::timestamp(arrow::TimeUnit::MICRO, timezone)), | |
| arrow::field("ts_tz_nano", arrow::timestamp(arrow::TimeUnit::NANO, timezone)), | |
| }; | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([ | |
| [0, "1970-01-01 00:00:01", "1970-01-01 00:00:00.001", "1970-01-01 00:00:00.000001", "1970-01-01 00:00:00.000000001", "1970-01-01 00:00:02", "1970-01-01 00:00:00.002", "1970-01-01 00:00:00.000002", "1970-01-01 00:00:00.000000002"], | |
| [0, "1970-01-01 00:00:03", "1970-01-01 00:00:00.003", null, "1970-01-01 00:00:00.000000003", "1970-01-01 00:00:04", "1970-01-01 00:00:00.004", "1970-01-01 00:00:00.000004", "1970-01-01 00:00:00.000000004"], | |
| [0, "1970-01-01 00:00:05", "1970-01-01 00:00:00.005", null, null, "1970-01-01 00:00:06", null, "1970-01-01 00:00:00.000006", null] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestCastTimestampType) { | |
| TimezoneGuard tz_guard("Asia/Shanghai"); | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/append_with_cast_timestamp.db" | |
| "/append_with_cast_timestamp/"; | |
| // scan | |
| ScanContextBuilder scan_context_builder(table_path); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 1); | |
| auto splits = result_plan->Splits(); | |
| ASSERT_EQ(1, splits.size()); | |
| // read | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(splits)); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result | |
| arrow::FieldVector fields = { | |
| arrow::field("_VALUE_KIND", arrow::int8()), | |
| arrow::field("ts_sec", arrow::int32()), | |
| arrow::field("ts_milli", arrow::date32()), | |
| arrow::field("ts_micro", arrow::utf8()), | |
| arrow::field("ts_nano", arrow::timestamp(arrow::TimeUnit::NANO, "Asia/Shanghai")), | |
| arrow::field("ts_tz_sec", arrow::int32()), | |
| arrow::field("ts_tz_milli", arrow::date32()), | |
| arrow::field("ts_tz_micro", arrow::utf8()), | |
| arrow::field("ts_tz_nano", arrow::timestamp(arrow::TimeUnit::NANO)), | |
| arrow::field("int_ts_sec", arrow::timestamp(arrow::TimeUnit::SECOND)), | |
| arrow::field("int_ts_tz_micro", arrow::timestamp(arrow::TimeUnit::MICRO, "Asia/Shanghai")), | |
| }; | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([ | |
| [0, -28799, 0, "1970-01-01 00:00:00.000001", "1969-12-31 16:00:00.000000001", 2, 0, "1970-01-01 08:00:00.000002", "1970-01-01 08:00:00.000000002", "1970-01-01 08:00:00", "1970-01-01 00:00:00.000000"], | |
| [0, -28797, 0, null, "1969-12-31 16:00:00.000000003", 4, 0, "1970-01-01 08:00:00.000004", "1970-01-01 08:00:00.000000004", "1970-01-01 08:00:01", "1970-01-01 00:00:01.000000"], | |
| [0, -28795, 0, null, null, 6, null, "1970-01-01 08:00:00.000006", null, "1970-01-01 07:59:59", "1969-12-31 23:59:59.000000"] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| } | |
| TEST_F(ScanAndReadInteTest, TestParquetDictionaryBinaryStoredArrowSchema) { | |
| const std::string table_path = GetDataDir() + | |
| "/parquet/append_dictionary_binary.db/" | |
| "append_dictionary_binary"; | |
| ScanContextBuilder scan_context_builder(table_path); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ScanContext> scan_context, scan_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableScan> table_scan, | |
| TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(1, plan->SnapshotId().value()); | |
| ASSERT_EQ(1, plan->Splits().size()); | |
| ReadContextBuilder read_context_builder(table_path); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableRead> table_read, | |
| TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<BatchReader> reader, | |
| table_read->CreateReader(plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> actual, | |
| ReadResultCollector::CollectResult(std::move(reader))); | |
| std::shared_ptr<arrow::DataType> expected_type = arrow::struct_({ | |
| arrow::field("_VALUE_KIND", arrow::int8()), | |
| arrow::field("payload", arrow::binary()), | |
| }); | |
| std::shared_ptr<arrow::Array> expected_array = | |
| arrow::ipc::internal::json::ArrayFromJSON( | |
| expected_type, R"([[0, "alpha"], [0, null], [0, "beta"], [0, "alpha"]])") | |
| .ValueOrDie(); | |
| std::shared_ptr<arrow::ChunkedArray> expected = | |
| std::make_shared<arrow::ChunkedArray>(expected_array); | |
| ASSERT_TRUE(expected->Equals(actual)) << actual->ToString(); | |
| } | |
| #ifdef PAIMON_ENABLE_LANCE | |
| TEST_F(ScanAndReadInteTest, TestLanceJavaCompatibility) { | |
| TimezoneGuard timezone_guard("UTC"); | |
| auto nested_type = arrow::struct_( | |
| {arrow::field("number", arrow::int32()), arrow::field("label", arrow::utf8())}); | |
| arrow::FieldVector fields = { | |
| arrow::field("_VALUE_KIND", arrow::int8()), | |
| arrow::field("id", arrow::int32()), | |
| arrow::field("f_boolean", arrow::boolean()), | |
| arrow::field("f_tinyint", arrow::int8()), | |
| arrow::field("f_smallint", arrow::int16()), | |
| arrow::field("f_bigint", arrow::int64()), | |
| arrow::field("f_float", arrow::float32()), | |
| arrow::field("f_double", arrow::float64()), | |
| arrow::field("f_char", arrow::utf8()), | |
| arrow::field("f_string", arrow::utf8()), | |
| arrow::field("f_binary", arrow::binary()), | |
| arrow::field("f_varbinary", arrow::binary()), | |
| arrow::field("f_date", arrow::date32()), | |
| arrow::field("f_time_0", arrow::time32(arrow::TimeUnit::MILLI)), | |
| arrow::field("f_time_3", arrow::time32(arrow::TimeUnit::MILLI)), | |
| arrow::field("f_ts_0", arrow::timestamp(arrow::TimeUnit::SECOND)), | |
| arrow::field("f_ts_3", arrow::timestamp(arrow::TimeUnit::MILLI)), | |
| arrow::field("f_ts_6", arrow::timestamp(arrow::TimeUnit::MICRO)), | |
| arrow::field("f_ts_9", arrow::timestamp(arrow::TimeUnit::NANO)), | |
| arrow::field("f_decimal_1_0", arrow::decimal128(1, 0)), | |
| arrow::field("f_decimal_18_2", arrow::decimal128(18, 2)), | |
| arrow::field("f_decimal_19_2", arrow::decimal128(19, 2)), | |
| arrow::field("f_decimal_38_18", arrow::decimal128(38, 18)), | |
| arrow::field("f_array_int", arrow::list(arrow::int32())), | |
| arrow::field("f_array_array_int", arrow::list(arrow::list(arrow::int32()))), | |
| arrow::field("f_struct", nested_type), | |
| arrow::field("f_nullable_struct", nested_type), | |
| arrow::field("f_vector", arrow::fixed_size_list(arrow::float32(), 3)), | |
| }; | |
| // Nullable-declared ROW is readable when every parent is valid. Null children are distinct | |
| // from null parents; see the README alongside the Java compatibility table. | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([ | |
| [0, 1, true, -5, -1000, 10000000001, 1.25, -2.5, "char0001", "value-1", "bin1", "\u0000\u0001\u0002\u007f", -1, 0, 86399999, "1970-01-01 00:00:01", "1970-01-01 00:00:01.123", "1970-01-01 00:00:01.123456", "1970-01-01 00:00:01.123456789", "-3", "1.25", "12345678901234567.89", "12345678901234567890.123456789012345678", [1, null, -1], [[1, null], null, []], [1, "required"], [1, "nullable"], [1.0, -1.5, 0.25]], | |
| [0, 2, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, [null, null], [null, null], null], | |
| [0, 10, false, -5, -1000, 10000000010, 1.25, -2.5, "char0001", "", "bin1", "\u0000\u0001\u0002\u007f", 20000, 45296000, 1123, "1970-01-01 00:00:01", "1970-01-01 00:00:01.123", "1970-01-01 00:00:01.123456", "1970-01-01 00:00:01.123456789", "-3", "1.25", "12345678901234567.89", "12345678901234567890.123456789012345678", [], [[10, null], null, []], [10, "required"], [10, "nullable"], [10.0, -1.5, 0.25]], | |
| [0, 11, true, -5, -1000, 10000000011, 1.25, -2.5, "char0001", "value-11", "bin1", "\u0000\u0001\u0002\u007f", 20000, 86399000, 45296789, "1970-01-01 00:00:01", "1970-01-01 00:00:01.123", "1970-01-01 00:00:01.123456", "1970-01-01 00:00:01.123456789", "-3", "1.25", "12345678901234567.89", "12345678901234567890.123456789012345678", [11, null, -11], [[11, null], null, []], [11, "required"], [11, "nullable"], [11.0, -1.5, 0.25]] | |
| ])") | |
| .ValueOrDie()); | |
| const std::string table_path = GetDataDir() + "/lance/append_java_compat.db/append_java_compat"; | |
| auto check = [&](const std::shared_ptr<Predicate>& predicate, | |
| const std::vector<std::string>& projection, | |
| const std::shared_ptr<arrow::ChunkedArray>& expected_result, | |
| bool exact_filter = false) { | |
| ScanContextBuilder scan_builder(table_path); | |
| ReadContextBuilder read_builder(table_path); | |
| if (predicate) { | |
| scan_builder.SetPredicate(predicate); | |
| if (projection.empty()) { | |
| read_builder.SetPredicate(predicate); | |
| } | |
| } | |
| if (!projection.empty()) { | |
| // The scan predicate uses table field indexes, not reordered projection indexes. | |
| read_builder.SetReadFieldNames(projection); | |
| } | |
| read_builder.EnablePredicateFilter(exact_filter); | |
| read_builder.AddOption("read.batch-size", "1"); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, scan_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto scan, TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(auto plan, scan->CreatePlan()); | |
| ASSERT_EQ(plan->SnapshotId(), std::optional<int64_t>(2)); | |
| ASSERT_FALSE(plan->Splits().empty()); | |
| ASSERT_OK_AND_ASSIGN(auto read_context, read_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto reader, read->CreateReader(plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(auto actual, ReadResultCollector::CollectResult(std::move(reader))); | |
| ASSERT_TRUE(actual); | |
| ASSERT_TRUE(expected_result->Equals(actual)) | |
| << "actual: " << actual->ToString() << "\nexpected: " << expected_result->ToString(); | |
| }; | |
| check(/*predicate=*/nullptr, /*projection=*/{}, expected); | |
| auto predicate = PredicateBuilder::GreaterOrEqual(0, "id", FieldType::INT, Literal(10)); | |
| // Missing statistics retain both files; exact filtering is opt-in above the format layer. | |
| check(predicate, /*projection=*/{}, expected); | |
| check(predicate, /*projection=*/{}, expected->Slice(2, 2), /*exact_filter=*/true); | |
| auto predicate_without_stats = PredicateBuilder::GreaterOrEqual( | |
| 4, "f_bigint", FieldType::BIGINT, Literal(int64_t{10000000010LL})); | |
| check(predicate_without_stats, /*projection=*/{}, expected); | |
| auto projected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON( | |
| arrow::struct_({fields[0], fields[26], fields[1], fields[27]}), R"([ | |
| [0, [1, "nullable"], 1, [1.0, -1.5, 0.25]], | |
| [0, [null, null], 2, null], | |
| [0, [10, "nullable"], 10, [10.0, -1.5, 0.25]], | |
| [0, [11, "nullable"], 11, [11.0, -1.5, 0.25]] | |
| ])") | |
| .ValueOrDie()); | |
| check(predicate, {"f_nullable_struct", "id", "f_vector"}, projected); | |
| } | |
| TEST_F(ScanAndReadInteTest, TestLancePythonCompatibility) { | |
| TimezoneGuard timezone_guard("UTC"); | |
| arrow::FieldVector fields = { | |
| arrow::field("_VALUE_KIND", arrow::int8()), | |
| arrow::field("id", arrow::int32()), | |
| arrow::field("f_boolean", arrow::boolean()), | |
| arrow::field("f_tinyint", arrow::int8()), | |
| arrow::field("f_smallint", arrow::int16()), | |
| arrow::field("f_bigint", arrow::int64()), | |
| arrow::field("f_float", arrow::float32()), | |
| arrow::field("f_double", arrow::float64()), | |
| arrow::field("f_string", arrow::utf8()), | |
| arrow::field("f_binary", arrow::binary()), | |
| arrow::field("f_date", arrow::date32()), | |
| arrow::field("f_time_0", arrow::time32(arrow::TimeUnit::MILLI)), | |
| arrow::field("f_time_3", arrow::time32(arrow::TimeUnit::MILLI)), | |
| arrow::field("f_ts_0", arrow::timestamp(arrow::TimeUnit::SECOND)), | |
| arrow::field("f_ts_3", arrow::timestamp(arrow::TimeUnit::MILLI)), | |
| arrow::field("f_ts_6", arrow::timestamp(arrow::TimeUnit::MICRO)), | |
| arrow::field("f_ts_9", arrow::timestamp(arrow::TimeUnit::NANO)), | |
| arrow::field("f_decimal_1_0", arrow::decimal128(1, 0)), | |
| arrow::field("f_decimal_18_2", arrow::decimal128(18, 2)), | |
| arrow::field("f_decimal_19_2", arrow::decimal128(19, 2)), | |
| arrow::field("f_decimal_38_18", arrow::decimal128(38, 18)), | |
| arrow::field("f_array_int", arrow::list(arrow::int32())), | |
| arrow::field("f_array_array_int", arrow::list(arrow::list(arrow::int32()))), | |
| arrow::field("f_struct", arrow::struct_({arrow::field("number", arrow::int32()), | |
| arrow::field("label", arrow::utf8())})), | |
| arrow::field("f_vector", arrow::fixed_size_list(arrow::float32(), 3)), | |
| }; | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([ | |
| [0, 1, true, -5, -1000, 10000000001, 1.25, -2.5, "value-1", "\u0000\u0001\u0002\u007f", -1, 0, 86399999, "1970-01-01 00:00:01", "1970-01-01 00:00:01.123", "1970-01-01 00:00:01.123456", "1970-01-01 00:00:01.123456789", "-3", "1.25", "12345678901234567.89", "12345678901234567890.123456789012345678", [1, null, -1], [[1, null], null, []], [1, "row-1"], [1.0, -1.5, 0.25]], | |
| [0, 2, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, [null, null], null], | |
| [0, 10, false, -5, -1000, 10000000010, 1.25, -2.5, "", "\u0000\u0001\u0002\u007f", 20000, 45296000, 1123, "1970-01-01 00:00:01", "1970-01-01 00:00:01.123", "1970-01-01 00:00:01.123456", "1970-01-01 00:00:01.123456789", "-3", "1.25", "12345678901234567.89", "12345678901234567890.123456789012345678", [], [[10, null], null, []], [10, "row-10"], [10.0, -1.5, 0.25]], | |
| [0, 11, true, -5, -1000, 10000000011, 1.25, -2.5, "value-11", "\u0000\u0001\u0002\u007f", 20000, 86399000, 45296789, "1970-01-01 00:00:01", "1970-01-01 00:00:01.123", "1970-01-01 00:00:01.123456", "1970-01-01 00:00:01.123456789", "-3", "1.25", "12345678901234567.89", "12345678901234567890.123456789012345678", [11, null, -11], [[11, null], null, []], [11, "row-11"], [11.0, -1.5, 0.25]] | |
| ])") | |
| .ValueOrDie()); | |
| const std::string table_path = | |
| GetDataDir() + "/lance/append_python_compat.db/append_python_compat"; | |
| auto check = [&](int64_t snapshot_id, const std::vector<std::string>& projection, | |
| const std::shared_ptr<arrow::ChunkedArray>& expected_result) { | |
| SCOPED_TRACE("snapshot " + std::to_string(snapshot_id)); | |
| 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 scan, TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(auto plan, scan->CreatePlan()); | |
| ASSERT_EQ(plan->SnapshotId(), std::optional<int64_t>(snapshot_id)); | |
| ASSERT_FALSE(plan->Splits().empty()); | |
| ReadContextBuilder read_builder(table_path); | |
| read_builder.AddOption("read.batch-size", "1"); | |
| if (!projection.empty()) { | |
| read_builder.SetReadFieldNames(projection); | |
| } | |
| ASSERT_OK_AND_ASSIGN(auto read_context, read_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto reader, read->CreateReader(plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(auto actual, ReadResultCollector::CollectResult(std::move(reader))); | |
| ASSERT_TRUE(actual); | |
| ASSERT_TRUE(expected_result->Equals(actual)) | |
| << "actual: " << actual->ToString() << "\nexpected: " << expected_result->ToString(); | |
| }; | |
| // These are independently committed Paimon snapshots, not standalone Lance datasets. | |
| check(1, /*projection=*/{}, expected->Slice(0, 2)); | |
| check(2, /*projection=*/{}, expected); | |
| auto projected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON( | |
| arrow::struct_({fields[0], fields[23], fields[1], fields[24], fields[21]}), R"([ | |
| [0, [1, "row-1"], 1, [1.0, -1.5, 0.25], [1, null, -1]], | |
| [0, [null, null], 2, null, null], | |
| [0, [10, "row-10"], 10, [10.0, -1.5, 0.25], []], | |
| [0, [11, "row-11"], 11, [11.0, -1.5, 0.25], [11, null, -11]] | |
| ])") | |
| .ValueOrDie()); | |
| check(1, {"f_struct", "id", "f_vector", "f_array_int"}, projected->Slice(0, 2)); | |
| check(2, {"f_struct", "id", "f_vector", "f_array_int"}, projected); | |
| } | |
| TEST_F(ScanAndReadInteTest, TestLanceLtzCompatibility) { | |
| TimezoneGuard timezone_guard("Asia/Shanghai"); | |
| const std::string zone = DateTimeUtils::GetLocalTimezoneName(); | |
| arrow::FieldVector fields = { | |
| arrow::field("_VALUE_KIND", arrow::int8()), | |
| arrow::field("id", arrow::int32()), | |
| arrow::field("f_ltz_0", arrow::timestamp(arrow::TimeUnit::SECOND, zone)), | |
| arrow::field("f_ltz_3", arrow::timestamp(arrow::TimeUnit::MILLI, zone)), | |
| arrow::field("f_ltz_6", arrow::timestamp(arrow::TimeUnit::MICRO, zone)), | |
| arrow::field("f_ltz_9", arrow::timestamp(arrow::TimeUnit::NANO, zone)), | |
| arrow::field("f_ltz_array", arrow::list(arrow::timestamp(arrow::TimeUnit::MICRO, zone))), | |
| arrow::field( | |
| "f_ltz_struct", | |
| arrow::struct_({arrow::field("number", arrow::int32()), | |
| arrow::field("ts", arrow::timestamp(arrow::TimeUnit::MILLI, zone))})), | |
| }; | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([ | |
| [0, 1, "1970-01-01 00:00:01", "1970-01-01 00:00:01.123", "1970-01-01 00:00:01.123456", "1970-01-01 00:00:01.123456789", ["1970-01-01 00:00:01.123456", null, "1969-12-31 23:59:59.999999"], [1, "1970-01-01 00:00:01.123"]], | |
| [0, 2, null, null, null, null, null, [null, null]], | |
| [0, 10, "1969-12-31 00:00:00", "1969-12-31 23:59:59.999", "1969-12-31 23:59:59.999999", "1969-12-31 23:59:59.999999999", [], [10, "1969-12-31 23:59:59.999"]], | |
| [0, 11, "2024-10-04 12:34:56", "2024-10-04 12:34:56.123", "2024-10-04 12:34:56.123456", "2024-10-04 12:34:56.123456789", ["2024-10-04 12:34:56.123456"], [11, "2024-10-04 12:34:56.123"]] | |
| ])") | |
| .ValueOrDie()); | |
| auto projected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON( | |
| arrow::struct_({fields[0], fields[7], fields[1], fields[5], fields[6]}), R"([ | |
| [0, [1, "1970-01-01 00:00:01.123"], 1, "1970-01-01 00:00:01.123456789", ["1970-01-01 00:00:01.123456", null, "1969-12-31 23:59:59.999999"]], | |
| [0, [null, null], 2, null, null], | |
| [0, [10, "1969-12-31 23:59:59.999"], 10, "1969-12-31 23:59:59.999999999", []], | |
| [0, [11, "2024-10-04 12:34:56.123"], 11, "2024-10-04 12:34:56.123456789", ["2024-10-04 12:34:56.123456"]] | |
| ])") | |
| .ValueOrDie()); | |
| auto check = [&](const std::string& table_path, int64_t snapshot_id, | |
| const std::vector<std::string>& projection, | |
| const std::shared_ptr<arrow::ChunkedArray>& expected_result) { | |
| SCOPED_TRACE("snapshot " + std::to_string(snapshot_id)); | |
| 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 scan, TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(auto plan, scan->CreatePlan()); | |
| ASSERT_EQ(plan->SnapshotId(), std::optional<int64_t>(snapshot_id)); | |
| ASSERT_FALSE(plan->Splits().empty()); | |
| ReadContextBuilder read_builder(table_path); | |
| read_builder.AddOption("read.batch-size", "1"); | |
| if (!projection.empty()) { | |
| read_builder.SetReadFieldNames(projection); | |
| } | |
| ASSERT_OK_AND_ASSIGN(auto read_context, read_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto reader, read->CreateReader(plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(auto actual, ReadResultCollector::CollectResult(std::move(reader))); | |
| ASSERT_TRUE(actual); | |
| ASSERT_TRUE(expected_result->Equals(actual)) | |
| << "actual: " << actual->ToString() << "\nexpected: " << expected_result->ToString(); | |
| }; | |
| for (const std::string table_name : {"append_ltz_java_compat", "append_ltz_python_compat"}) { | |
| SCOPED_TRACE(table_name); | |
| std::string table_path = GetDataDir() + "/lance/" + table_name + ".db/" + table_name; | |
| check(table_path, 1, /*projection=*/{}, expected->Slice(0, 2)); | |
| check(table_path, 2, /*projection=*/{}, expected); | |
| check(table_path, 2, {"f_ltz_struct", "id", "f_ltz_9", "f_ltz_array"}, projected); | |
| } | |
| } | |
| #endif | |
| #ifdef PAIMON_ENABLE_MOSAIC | |
| TEST_F(ScanAndReadInteTest, TestMosaicJavaAndPythonCompatibility) { | |
| TimezoneGuard timezone_guard("UTC"); | |
| std::string timezone = DateTimeUtils::GetLocalTimezoneName(); | |
| arrow::FieldVector fields = { | |
| arrow::field("_VALUE_KIND", arrow::int8()), | |
| arrow::field("id", arrow::int32()), | |
| arrow::field("f_boolean", arrow::boolean()), | |
| arrow::field("f_tinyint", arrow::int8()), | |
| arrow::field("f_smallint", arrow::int16()), | |
| arrow::field("f_bigint", arrow::int64()), | |
| arrow::field("f_float", arrow::float32()), | |
| arrow::field("f_double", arrow::float64()), | |
| arrow::field("f_char", arrow::utf8()), | |
| arrow::field("f_varchar", arrow::utf8()), | |
| arrow::field("f_binary", arrow::binary()), | |
| arrow::field("f_varbinary", arrow::binary()), | |
| arrow::field("f_date", arrow::date32()), | |
| arrow::field("f_ts_3", arrow::timestamp(arrow::TimeUnit::MILLI)), | |
| arrow::field("f_ts_6", arrow::timestamp(arrow::TimeUnit::MICRO)), | |
| arrow::field("f_ts_9", arrow::timestamp(arrow::TimeUnit::NANO)), | |
| arrow::field("f_ltz_3", arrow::timestamp(arrow::TimeUnit::MILLI, timezone)), | |
| arrow::field("f_ltz_6", arrow::timestamp(arrow::TimeUnit::MICRO, timezone)), | |
| arrow::field("f_ltz_9", arrow::timestamp(arrow::TimeUnit::NANO, timezone)), | |
| arrow::field("f_decimal_1_0", arrow::decimal128(1, 0)), | |
| arrow::field("f_decimal_18_2", arrow::decimal128(18, 2)), | |
| arrow::field("f_decimal_19_2", arrow::decimal128(19, 2)), | |
| arrow::field("f_decimal_38_18", arrow::decimal128(38, 18)), | |
| arrow::field("f_array_int", arrow::list(arrow::int32())), | |
| arrow::field("f_map_numeric", arrow::map(arrow::int8(), arrow::int16())), | |
| arrow::field("f_map_string_bigint", arrow::map(arrow::utf8(), arrow::int64())), | |
| arrow::field("f_array_array_int", arrow::list(arrow::list(arrow::int32()))), | |
| arrow::field("f_array_map", arrow::list(arrow::map(arrow::utf8(), arrow::int32()))), | |
| arrow::field("f_map_array", arrow::map(arrow::utf8(), arrow::list(arrow::int32()))), | |
| }; | |
| std::shared_ptr<arrow::DataType> data_type = arrow::struct_(fields); | |
| std::shared_ptr<arrow::Array> expected_array = | |
| arrow::ipc::internal::json::ArrayFromJSON(data_type, R"([ | |
| [0, 1, true, -5, -1000, 10000000001, 1.25, 10.5, "char0001", "value-1", "bin00001", "\u0001\u0002\u0003", 20000, "1970-01-01 00:00:01.123", "1970-01-01 00:00:01.123456", "1970-01-01 00:00:01.123456789", "1970-01-01 00:01:01.321", "1970-01-01 00:01:01.654321", "1970-01-01 00:01:01.321654987", "-3", "1.25", "12345678901234567.89", "12345678901234567890.123456789012345678", [1, 2], [[0, 0], [10, 1]], [["k0", 1000], ["z0", 2000]], [[1, 2], [3]], [[["nested0", 0]], [["nested10", 10]]], [["array0", [0, 1]]]], | |
| [0, 2, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null], | |
| [0, 10, true, -3, -998, 10000000010, 3.25, 12.5, "char0010", "value-10", "bin00010", "\u000a\u000b\u000c", 20002, "1970-01-01 00:00:10.123", "1970-01-01 00:00:10.123456", "1970-01-01 00:00:10.123456789", "1970-01-01 00:01:10.321", "1970-01-01 00:01:10.654321", "1970-01-01 00:01:10.321654987", "-1", "10.25", "12345678901234569.89", "12345678901234567892.123456789012345678", [10, 11], [[2, 20], [12, 21]], [["k2", 1002], ["z2", 2002]], [[10, 11], [12]], [[["nested2", 2]], [["nested12", 12]]], [["array2", [2, 3]]]], | |
| [0, 11, false, -2, -997, 10000000011, 4.25, 13.5, "char0011", "value-11", "bin00011", "\u000b\u000c\u000d", 20003, "1970-01-01 00:00:11.123", "1970-01-01 00:00:11.123456", "1970-01-01 00:00:11.123456789", "1970-01-01 00:01:11.321", "1970-01-01 00:01:11.654321", "1970-01-01 00:01:11.321654987", "0", "11.25", "12345678901234570.89", "12345678901234567893.123456789012345678", [11, 12], [[3, 30], [13, 31]], [["k3", 1003], ["z3", 2003]], [[11, 12], [13]], [[["nested3", 3]], [["nested13", 13]]], [["array3", [3, 4]]]], | |
| [0, 20, true, -1, -996, 10000000020, 5.25, 14.5, "char0020", "value-20", "bin00020", "\u0014\u0015\u0016", 20004, "1970-01-01 00:00:20.123", "1970-01-01 00:00:20.123456", "1970-01-01 00:00:20.123456789", "1970-01-01 00:01:20.321", "1970-01-01 00:01:20.654321", "1970-01-01 00:01:20.321654987", "1", "20.25", "12345678901234571.89", "12345678901234567894.123456789012345678", [20, 21], [[4, 40], [14, 41]], [["k4", 1004], ["z4", 2004]], [[20, 21], [22]], [[["nested4", 4]], [["nested14", 14]]], [["array4", [4, 5]]]], | |
| [0, 21, false, 0, -995, 10000000021, 6.25, 15.5, "char0021", "value-21", "bin00021", "\u0015\u0016\u0017", 20005, "1970-01-01 00:00:21.123", "1970-01-01 00:00:21.123456", "1970-01-01 00:00:21.123456789", "1970-01-01 00:01:21.321", "1970-01-01 00:01:21.654321", "1970-01-01 00:01:21.321654987", "2", "21.25", "12345678901234572.89", "12345678901234567895.123456789012345678", [21, 22], [[5, 50], [15, 51]], [["k5", 1005], ["z5", 2005]], [[21, 22], [23]], [[["nested5", 5]], [["nested15", 15]]], [["array5", [5, 6]]]] | |
| ])") | |
| .ValueOrDie(); | |
| auto expected = std::make_shared<arrow::ChunkedArray>(expected_array); | |
| auto check_compatibility = [](const std::string& table_path, | |
| const std::shared_ptr<Predicate>& predicate, | |
| const std::shared_ptr<arrow::ChunkedArray>& expected_result) { | |
| ScanContextBuilder scan_context_builder(table_path); | |
| ReadContextBuilder read_context_builder(table_path); | |
| if (predicate != nullptr) { | |
| scan_context_builder.SetPredicate(predicate); | |
| read_context_builder.SetPredicate(predicate); | |
| } | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ScanContext> scan_context, | |
| scan_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableScan> table_scan, | |
| TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> plan, table_scan->CreatePlan()); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, | |
| read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableRead> table_read, | |
| TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<BatchReader> batch_reader, | |
| table_read->CreateReader(plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> actual, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| ASSERT_TRUE(expected_result->Equals(actual)) | |
| << "actual: " << (actual == nullptr ? "null" : actual->ToString()) | |
| << "\nexpected: " << expected_result->ToString(); | |
| }; | |
| std::shared_ptr<Predicate> predicate = PredicateBuilder::GreaterOrEqual( | |
| /*field_index=*/0, /*field_name=*/"id", FieldType::INT, Literal(20)); | |
| std::shared_ptr<Predicate> predicate_without_stats = PredicateBuilder::GreaterOrEqual( | |
| /*field_index=*/4, /*field_name=*/"f_bigint", FieldType::BIGINT, | |
| Literal(int64_t{10000000020LL})); | |
| for (const std::string table_name : {"append_java_compat", "append_python_compat"}) { | |
| SCOPED_TRACE(table_name); | |
| std::string table_path = GetDataDir() + "/mosaic/" + table_name + ".db/" + table_name; | |
| check_compatibility(table_path, /*predicate=*/nullptr, expected); | |
| check_compatibility(table_path, predicate, expected->Slice(4, 2)); | |
| check_compatibility(table_path, predicate_without_stats, expected); | |
| } | |
| } | |
| #endif | |
| #ifdef PAIMON_ENABLE_VORTEX | |
| TEST_F(ScanAndReadInteTest, TestVortexJavaCompatibility) { | |
| // Fixture written by Paimon Java 2.0.0 with file.format=vortex; see | |
| // test/test_data/vortex/append_java_compat.db/append_java_compat/README.md. This verifies that | |
| // paimon-cpp reads back a Vortex table produced by the Java implementation. | |
| arrow::FieldVector fields = { | |
| arrow::field("_VALUE_KIND", arrow::int8()), | |
| arrow::field("id", arrow::int32()), | |
| arrow::field("name", arrow::utf8()), | |
| }; | |
| std::shared_ptr<arrow::DataType> data_type = arrow::struct_(fields); | |
| std::shared_ptr<arrow::Array> expected_array = | |
| arrow::ipc::internal::json::ArrayFromJSON(data_type, R"([ | |
| [0, 1, "one"], | |
| [0, 2, null], | |
| [0, 3, "three"], | |
| [0, 4, "four"], | |
| [0, 5, "five"] | |
| ])") | |
| .ValueOrDie(); | |
| auto expected = std::make_shared<arrow::ChunkedArray>(expected_array); | |
| const std::string table_name = "append_java_compat"; | |
| const std::string table_path = GetDataDir() + "/vortex/" + table_name + ".db/" + table_name; | |
| ScanContextBuilder scan_context_builder(table_path); | |
| ReadContextBuilder read_context_builder(table_path); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ScanContext> scan_context, scan_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableScan> table_scan, | |
| TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> plan, table_scan->CreatePlan()); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableRead> table_read, | |
| TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<BatchReader> batch_reader, | |
| table_read->CreateReader(plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> actual, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| ASSERT_TRUE(expected->Equals(actual)) | |
| << "actual: " << (actual == nullptr ? "null" : actual->ToString()) | |
| << "\nexpected: " << expected->ToString(); | |
| } | |
| // The Java fixture stores TIMESTAMP_LTZ in UTC while paimon reads it in the reader's own zone, so | |
| // the same file must yield the same instants under any timezone. | |
| TEST_F(ScanAndReadInteTest, TestVortexJavaCompatibilityLtzAcrossTimezones) { | |
| const std::string table_path = | |
| GetDataDir() + "/vortex/all_types_java_compat.db/all_types_java_compat"; | |
| std::shared_ptr<arrow::Array> utc; | |
| std::shared_ptr<arrow::Array> shanghai; | |
| const std::string timezones[] = {"UTC", "Asia/Shanghai"}; | |
| for (const std::string& tz : timezones) { | |
| TimezoneGuard timezone_guard(tz); | |
| ScanContextBuilder scan_context_builder(table_path); | |
| ReadContextBuilder read_context_builder(table_path); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ScanContext> scan_context, | |
| scan_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableScan> table_scan, | |
| TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> plan, table_scan->CreatePlan()); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, | |
| read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableRead> table_read, | |
| TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<BatchReader> batch_reader, | |
| table_read->CreateReader(plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> actual, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| ASSERT_EQ(actual->num_chunks(), 1); | |
| const auto& row = checked_cast<const arrow::StructArray&>(*actual->chunk(0)); | |
| std::shared_ptr<arrow::Array> column = row.GetFieldByName("f_ltz_3"); | |
| ASSERT_NE(column, nullptr); | |
| ASSERT_EQ(column->type()->ToString(), "timestamp[ms, tz=" + tz + "]"); | |
| if (tz == "UTC") { | |
| utc = column; | |
| } else { | |
| shanghai = column; | |
| } | |
| } | |
| arrow::Result<std::shared_ptr<arrow::Array>> viewed = shanghai->View(utc->type()); | |
| ASSERT_TRUE(viewed.ok()) << viewed.status().ToString(); | |
| ASSERT_TRUE((*viewed)->Equals(*utc)) | |
| << "shanghai: " << shanghai->ToString() << "\nutc: " << utc->ToString(); | |
| } | |
| TEST_F(ScanAndReadInteTest, TestVortexJavaCompatibilityAllTypes) { | |
| // Fixture written by Paimon Java 2.0.0 with file.format=vortex; see | |
| // test/test_data/vortex/all_types_java_compat.db/all_types_java_compat/README.md. Covers every | |
| // field type paimon-cpp accepts for the Vortex format, including an all-null row. | |
| TimezoneGuard timezone_guard("UTC"); | |
| std::string timezone = DateTimeUtils::GetLocalTimezoneName(); | |
| arrow::FieldVector fields = { | |
| arrow::field("_VALUE_KIND", arrow::int8()), | |
| arrow::field("f_boolean", arrow::boolean()), | |
| arrow::field("f_tinyint", arrow::int8()), | |
| arrow::field("f_smallint", arrow::int16()), | |
| arrow::field("f_int", arrow::int32()), | |
| arrow::field("f_bigint", arrow::int64()), | |
| arrow::field("f_float", arrow::float32()), | |
| arrow::field("f_double", arrow::float64()), | |
| arrow::field("f_char", arrow::utf8()), | |
| arrow::field("f_varchar", arrow::utf8()), | |
| arrow::field("f_binary", arrow::binary()), | |
| arrow::field("f_varbinary", arrow::binary()), | |
| arrow::field("f_date", arrow::date32()), | |
| arrow::field("f_time", arrow::time32(arrow::TimeUnit::MILLI)), | |
| arrow::field("f_ts_3", arrow::timestamp(arrow::TimeUnit::MILLI)), | |
| arrow::field("f_ts_6", arrow::timestamp(arrow::TimeUnit::MICRO)), | |
| arrow::field("f_ltz_3", arrow::timestamp(arrow::TimeUnit::MILLI, timezone)), | |
| arrow::field("f_decimal_10_2", arrow::decimal128(10, 2)), | |
| arrow::field("f_decimal_38_18", arrow::decimal128(38, 18)), | |
| arrow::field("f_array_int", arrow::list(arrow::int32())), | |
| arrow::field("f_array_array_int", arrow::list(arrow::list(arrow::int32()))), | |
| arrow::field("f_vector_float", arrow::fixed_size_list(arrow::float32(), 3)), | |
| arrow::field("f_row", arrow::struct_({arrow::field("a", arrow::int32()), | |
| arrow::field("b", arrow::utf8())}))}; | |
| std::shared_ptr<arrow::DataType> data_type = arrow::struct_(fields); | |
| std::shared_ptr<arrow::Array> expected_array = | |
| arrow::ipc::internal::json::ArrayFromJSON(data_type, R"([ | |
| [0, true, -5, -1000, 1, 10000000001, 1.25, 10.5, "char0001", "value-1", "\u0001\u0002\u0003\u0004\u0005\u0006", "\u0004\u0005", 20000, 3723123, "2023-11-14 22:13:21.123", "2023-11-14 22:13:21.123456", "2023-11-14 22:14:21.321", "1.25", "12345678901234567890.123456789012345678", [1, 2], [[1, 2], [3]], [1.0, 2.0, 3.0], {"a": 1, "b": "x"}], | |
| [0, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null], | |
| [0, false, -3, -998, 10, 10000000010, 3.25, 12.5, "char0010", "value-10", "\u000a\u000b\u000c\u000d\u000e\u000f", "\u0006\u0007", 20002, 3733123, "2023-11-14 22:13:31.123", "2023-11-14 22:13:31.123654", "2023-11-14 22:14:31.321", "10.25", "12345678901234567892.123456789012345678", [10, 11], [[10, 11], [12]], [4.0, 5.0, 6.0], {"a": 2, "b": "y"}] | |
| ])") | |
| .ValueOrDie(); | |
| auto expected = std::make_shared<arrow::ChunkedArray>(expected_array); | |
| const std::string table_name = "all_types_java_compat"; | |
| const std::string table_path = GetDataDir() + "/vortex/" + table_name + ".db/" + table_name; | |
| ScanContextBuilder scan_context_builder(table_path); | |
| ReadContextBuilder read_context_builder(table_path); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ScanContext> scan_context, scan_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableScan> table_scan, | |
| TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> plan, table_scan->CreatePlan()); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableRead> table_read, | |
| TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<BatchReader> batch_reader, | |
| table_read->CreateReader(plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> actual, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| ASSERT_TRUE(expected->Equals(actual)) | |
| << "actual: " << (actual == nullptr ? "null" : actual->ToString()) | |
| << "\nexpected: " << expected->ToString(); | |
| } | |
| #endif | |
| TEST_F(ScanAndReadInteTest, TestAvroWithAppendTable) { | |
| auto read_data = [](int64_t snapshot_id, const std::string& result_json) { | |
| std::string table_path = GetDataDir() + "/avro/append_multiple.db/append_multiple"; | |
| // scan | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, std::to_string(snapshot_id)); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, scan_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), snapshot_id); | |
| auto splits = result_plan->Splits(); | |
| ASSERT_EQ(3, splits.size()); | |
| // read | |
| ReadContextBuilder read_context_builder(table_path); | |
| read_context_builder.AddOption("test.enable-adaptive-prefetch-strategy", "false"); | |
| read_context_builder.EnablePrefetch(true).SetPrefetchBatchCount(3); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, | |
| read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(splits)); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result | |
| auto timezone = DateTimeUtils::GetLocalTimezoneName(); | |
| arrow::FieldVector fields = { | |
| arrow::field("_VALUE_KIND", arrow::int8()), | |
| arrow::field("f0", arrow::int8()), | |
| arrow::field("f1", arrow::int16()), | |
| arrow::field("f2", arrow::int32()), | |
| arrow::field("f3", arrow::int64()), | |
| arrow::field("f4", arrow::float32()), | |
| arrow::field("f5", arrow::float64()), | |
| arrow::field("f6", arrow::utf8()), | |
| arrow::field("f7", arrow::binary()), | |
| arrow::field("f8", arrow::date32()), | |
| arrow::field("f9", arrow::decimal128(5, 2)), | |
| arrow::field("f10", arrow::timestamp(arrow::TimeUnit::SECOND)), | |
| arrow::field("f11", arrow::timestamp(arrow::TimeUnit::MILLI)), | |
| arrow::field("f12", arrow::timestamp(arrow::TimeUnit::MICRO)), | |
| arrow::field("f13", arrow::timestamp(arrow::TimeUnit::SECOND, timezone)), | |
| arrow::field("f14", arrow::timestamp(arrow::TimeUnit::MILLI, timezone)), | |
| arrow::field("f15", arrow::timestamp(arrow::TimeUnit::MICRO, timezone)), | |
| arrow::field("f16", arrow::struct_( | |
| {arrow::field("f0", arrow::map(arrow::utf8(), arrow::int32())), | |
| arrow::field("f1", arrow::list(arrow::int32()))})), | |
| }; | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), result_json) | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) | |
| << "read_result: " << read_result->ToString() << "expected: " << expected->ToString(); | |
| }; | |
| read_data(1, R"([ | |
| [0, 2, 10, 1, 100, 2.0, 2.0, "two", "bbb", 123, "123.45", "1970-01-02 00:00:00", "1970-01-02 00:00:00.000", "1970-01-02 00:00:00.000000", "1970-01-02 00:00:00", "1970-01-02 00:00:00.000", "1970-01-02 00:00:00.000000",[[["key",123]],[1,2,3]]], | |
| [0, 1, 10, 0, 100, 1.0, 1.0, "one", "aaa", 123, "123.45", "1970-01-01 00:00:00", "1970-01-01 00:00:00.000", "1970-01-01 00:00:00.000000", "1970-01-01 00:00:00", "1970-01-01 00:00:00.000", "1970-01-01 00:00:00.000000",[[["key",123]],[1,2,3]]], | |
| [0, 3, 11, 0, 100, null, 3.0, "three", "ccc", 123, "123.45", "1970-01-03 00:00:00", "1970-01-03 00:00:00.000", "1970-01-03 00:00:00.000000", "1970-01-03 00:00:00", "1970-01-03 00:00:00.000", "1970-01-03 00:00:00.000000",[[["key",123]],[1,2,3]]], | |
| [0, 4, 11, 0, 100, 4.0, null, "four", "ddd", 123, "123.45", "1970-01-04 00:00:00", "1970-01-04 00:00:00.000", "1970-01-04 00:00:00.000000", "1970-01-04 00:00:00", "1970-01-04 00:00:00.000", "1970-01-04 00:00:00.000000",[[["key",123]],[1,2,3]]] | |
| ])"); | |
| read_data(2, R"([ | |
| [0, 2, 10, 1, 100, 2.0, 2.0, "two", "bbb", 123, "123.45", "1970-01-02 00:00:00", "1970-01-02 00:00:00.000", "1970-01-02 00:00:00.000000", "1970-01-02 00:00:00", "1970-01-02 00:00:00.000", "1970-01-02 00:00:00.000000",[[["key",123]],[1,2,3]]], | |
| [0, 6, 10, 1, 100, 6.0, 4.0, "six", "fff", 123, "123.45", "1970-01-06 00:00:00", "1970-01-06 00:00:00.000", "1970-01-06 00:00:00.000000", "1970-01-06 00:00:00", "1970-01-06 00:00:00.000", "1970-01-06 00:00:00.000000",[[["key",123]],[1,2,3]]], | |
| [0, 1, 10, 0, 100, 1.0, 1.0, "one", "aaa", 123, "123.45", "1970-01-01 00:00:00", "1970-01-01 00:00:00.000", "1970-01-01 00:00:00.000000", "1970-01-01 00:00:00", "1970-01-01 00:00:00.000", "1970-01-01 00:00:00.000000",[[["key",123]],[1,2,3]]], | |
| [0, 5, 10, 0, 100, 5.0, 2.0, null, "eee", 123, "123.45", "1970-01-05 00:00:00", "1970-01-05 00:00:00.000", "1970-01-05 00:00:00.000000", "1970-01-05 00:00:00", "1970-01-05 00:00:00.000", "1970-01-05 00:00:00.000000",[[["key",123]],[1,2,3]]], | |
| [0, 3, 11, 0, 100, null, 3.0, "three", "ccc", 123, "123.45", "1970-01-03 00:00:00", "1970-01-03 00:00:00.000", "1970-01-03 00:00:00.000000", "1970-01-03 00:00:00", "1970-01-03 00:00:00.000", "1970-01-03 00:00:00.000000",[[["key",123]],[1,2,3]]], | |
| [0, 4, 11, 0, 100, 4.0, null, "four", "ddd", 123, "123.45", "1970-01-04 00:00:00", "1970-01-04 00:00:00.000", "1970-01-04 00:00:00.000000", "1970-01-04 00:00:00", "1970-01-04 00:00:00.000", "1970-01-04 00:00:00.000000",[[["key",123]],[1,2,3]]], | |
| [0, 7, 11, 0, 100, 7.0, 6.0, "seven", "ggg", 123, "123.45", "1970-01-07 00:00:00", "1970-01-07 00:00:00.000", "1970-01-07 00:00:00.000000", "1970-01-07 00:00:00", "1970-01-07 00:00:00.000", "1970-01-07 00:00:00.000000",[[["key",123]],[1,2,3]]] | |
| ])"); | |
| } | |
| TEST_F(ScanAndReadInteTest, TestAvroWithPkTable) { | |
| auto read_data = [](int64_t snapshot_id, const std::string& result_json) { | |
| std::string table_path = | |
| GetDataDir() + "/avro/pk_with_multiple_type.db/pk_with_multiple_type"; | |
| // scan | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, std::to_string(snapshot_id)); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, scan_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), snapshot_id); | |
| auto splits = result_plan->Splits(); | |
| ASSERT_EQ(1, splits.size()); | |
| // read | |
| ReadContextBuilder read_context_builder(table_path); | |
| ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, | |
| read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(splits)); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // check result | |
| arrow::FieldVector fields = { | |
| arrow::field("_VALUE_KIND", arrow::int8()), | |
| arrow::field("f0", arrow::boolean()), | |
| arrow::field("f1", arrow::int8()), | |
| arrow::field("f2", arrow::int16()), | |
| arrow::field("f3", arrow::int32()), | |
| arrow::field("f4", arrow::int64()), | |
| arrow::field("f5", arrow::float32()), | |
| arrow::field("f6", arrow::float64()), | |
| arrow::field("f7", arrow::utf8()), | |
| arrow::field("f8", arrow::binary()), | |
| arrow::field("f9", arrow::date32()), | |
| arrow::field("f10", arrow::decimal128(5, 2)), | |
| arrow::field("f11", arrow::struct_( | |
| {arrow::field("f0", arrow::map(arrow::utf8(), arrow::int32())), | |
| arrow::field("f1", arrow::list(arrow::int32()))})), | |
| }; | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(struct_(fields), result_json).ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| }; | |
| read_data(1, R"([ | |
| [0, false, 10, 1, 1, 1000, 1.5, 2.5, "Alice", "abcdef", 100, "123.45", [[["key",123]],[1,2,3]]], | |
| [0, false, 10, 1, 1, 1000, 1.5, 2.5, "Bob", "abcdef", 100, "123.45", [[["key",123]],[1,2,3]]], | |
| [0, true, 10, 1, 1, 1000, 1.5, 2.5, "Emily", "abcdef", 100, "123.45", [[["key",123]],[1,2,3]]], | |
| [0, true, 10, 1, 1, 1000, 1.5, 2.5, "Tony", "abcdef", 100, "123.45", [[["key",123]],[1,2,3]]] | |
| ])"); | |
| read_data(2, R"([ | |
| [0, false, 10, 1, 1, 1000, 1.5, 2.5, "Alice", "abcdef", 100, "123.45", [[["key",123]],[1,2,3]]], | |
| [0, false, 10, 1, 1, 1000, 1.5, 2.5, "Bob", "abcdef", 100, "123.45", [[["key",123]],[1,2,3]]], | |
| [0, true, 10, 1, 1, 1000, 1.5, 2.5, "Lucy", "abcdef", 100, "123.45", [[["key",123]],[1,2,3]]], | |
| [0, true, 10, 1, 1, 1000, 1.5, 2.5, "Tony", "abcdef", 100, "123.45", [[["key",123]],[1,2,3]]] | |
| ])"); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestWithPKBucketSelectByPredicate) { | |
| auto file_format = FileFormat(); | |
| // Verify BucketSelectConverter: an EQUAL predicate on bucket key f2 should automatically | |
| // derive the target bucket, without explicitly calling SetBucketFilter. | |
| // From the existing test f2=0 maps to bucket 1, f2=1 maps to bucket 0. | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_dv.db/pk_table_scan_and_read_dv/"; | |
| // Predicate: f2 = 0 (bucket key). This should cause BucketSelectConverter to derive bucket 1. | |
| auto predicate = PredicateBuilder::Equal(/*field_index=*/2, /*field_name=*/"f2", FieldType::INT, | |
| Literal(static_cast<int32_t>(0))); | |
| // Historical files use two buckets even when the current option has changed. | |
| for (const std::string current_bucket_count : {"2", "4", "8", "17"}) { | |
| SCOPED_TRACE(current_bucket_count); | |
| ScanContextBuilder scan_context_builder(table_path); | |
| scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "6"); | |
| scan_context_builder.AddOption(Options::BUCKET, current_bucket_count); | |
| scan_context_builder.SetPartitionFilter({{{"f1", "10"}}}); | |
| scan_context_builder.SetPredicate(predicate); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| read_context_builder.SetPredicate(predicate); | |
| ASSERT_OK_AND_ASSIGN(auto read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| if (EnableSnapshotLiveManifestCache()) { | |
| ASSERT_OK_AND_ASSIGN(uint64_t enabled, table_scan->GetMetrics()->GetCounter( | |
| ScanMetrics::LAST_SNAPSHOT_CACHE_ENABLED)); | |
| ASSERT_EQ(enabled, 1); | |
| ScanContextBuilder cached_builder(table_path); | |
| cached_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "6") | |
| .AddOption(Options::BUCKET, current_bucket_count) | |
| .SetPartitionFilter({{{"f1", "10"}}}) | |
| .SetPredicate(predicate); | |
| ASSERT_OK_AND_ASSIGN(auto cached_context, FinishScanContext(cached_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto cached_scan, TableScan::Create(std::move(cached_context))); | |
| ASSERT_OK_AND_ASSIGN(result_plan, cached_scan->CreatePlan()); | |
| ASSERT_OK_AND_ASSIGN(uint64_t hit, cached_scan->GetMetrics()->GetCounter( | |
| ScanMetrics::LAST_SNAPSHOT_CACHE_HIT)); | |
| ASSERT_EQ(hit, 1); | |
| } | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 6); | |
| // Verify all returned splits are from bucket 1 (f2=0 hashes to bucket 1) | |
| auto splits = result_plan->Splits(); | |
| ASSERT_FALSE(splits.empty()); | |
| for (const auto& split : splits) { | |
| auto data_split = std::dynamic_pointer_cast<DataSplitImpl>(split); | |
| ASSERT_TRUE(data_split); | |
| ASSERT_EQ(data_split->Bucket(), 1); | |
| } | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(splits)); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| // Only rows with f2=0 in partition f1=10 should be returned | |
| auto expected = std::make_shared<arrow::ChunkedArray>( | |
| arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type_, R"([ | |
| [0, "Alex", 10, 0, 16.1], | |
| [0, "Bob", 10, 0, 12.1], | |
| [0, "David", 10, 0, 17.1], | |
| [0, "Emily", 10, 0, 13.1] | |
| ])") | |
| .ValueOrDie()); | |
| ASSERT_TRUE(expected); | |
| ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); | |
| } | |
| } | |
| TEST_P(ScanAndReadInteTest, TestReadNullableMapKey) { | |
| auto file_format = FileFormat(); | |
| if (file_format != "orc") { | |
| return; | |
| } | |
| // Java Parquet does not support writing null MAP keys; see | |
| // ParquetRowDataWriter.MapWriter#writeMapData in Apache Paimon. Therefore, this test uses an | |
| // ORC table. The table contains three rows: the first two rows have non-null MAP keys, while | |
| // the third row has a null MAP key and is expected to fail during reading. | |
| const std::string table_path = | |
| paimon::test::GetDataDir() + "orc/nullable_map_key.db/nullable_map_key"; | |
| ScanContextBuilder scan_context_builder(table_path); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, scan_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ASSERT_TRUE(result_plan->SnapshotId()); | |
| ASSERT_EQ(result_plan->SnapshotId().value(), 1); | |
| ASSERT_EQ(result_plan->Splits().size(), 1); | |
| ReadContextBuilder read_context_builder(table_path); | |
| read_context_builder.AddOption(Options::READ_BATCH_SIZE, "1"); | |
| ASSERT_OK_AND_ASSIGN(auto read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(result_plan->Splits())); | |
| const std::vector<std::string> expected_rows = {R"([[0, 1, [["one", 10]]]])", | |
| R"([[0, 2, [["two", 20]]]])"}; | |
| std::vector<BatchReader::ReadBatch> batches; | |
| for (int32_t i = 0; i < 2; ++i) { | |
| ASSERT_OK_AND_ASSIGN(BatchReader::ReadBatch batch, batch_reader->NextBatch()); | |
| ASSERT_FALSE(BatchReader::IsEofBatch(batch)); | |
| batches.push_back(std::move(batch)); | |
| } | |
| ASSERT_NOK_WITH_MSG(batch_reader->NextBatch(), "Map array keys array should have no nulls"); | |
| batch_reader->Close(); | |
| batch_reader.reset(); | |
| for (int32_t i = 0; i < 2; ++i) { | |
| ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> array, | |
| ReadResultCollector::GetArray(std::move(batches[i]))); | |
| ASSERT_EQ(array->length(), 1); | |
| std::shared_ptr<arrow::Array> expected_array = | |
| arrow::ipc::internal::json::ArrayFromJSON(array->type(), expected_rows[i]).ValueOrDie(); | |
| ASSERT_TRUE(array->Equals(expected_array)) | |
| << "actual: " << array->ToString() << ", expected: " << expected_array->ToString(); | |
| } | |
| } | |
| TEST_P(ScanAndReadInteTest, TestCountRowsEmptySplits) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_mor.db/pk_table_scan_and_read_mor/"; | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(auto read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| // CreateCountReader with empty splits should return 0 rows. | |
| std::vector<std::shared_ptr<Split>> empty_splits; | |
| ASSERT_OK_AND_ASSIGN(auto count_reader, table_read->CreateCountReader(empty_splits)); | |
| ASSERT_OK_AND_ASSIGN(int64_t count, count_reader->CountRows()); | |
| ASSERT_EQ(count, 0); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestCountRowsConsistencyWithCreateReader) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_mor.db/pk_table_scan_and_read_mor/"; | |
| // Scan latest snapshot | |
| ScanContextBuilder scan_context_builder(table_path); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| // Method 1: CreateCountReader + iterate batches | |
| ReadContextBuilder count_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&count_context_builder); | |
| ASSERT_OK_AND_ASSIGN(auto count_read_context, count_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto count_table_read, TableRead::Create(std::move(count_read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto count_reader, | |
| count_table_read->CreateCountReader(result_plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(int64_t count_result, count_reader->CountRows()); | |
| // Method 2: CreateReader + iterate batches | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(auto read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(result_plan->Splits())); | |
| ASSERT_OK_AND_ASSIGN(auto read_result, | |
| ReadResultCollector::CollectResult(std::move(batch_reader))); | |
| int64_t iterate_count = read_result ? read_result->length() : 0; | |
| // Both methods should return the same count | |
| ASSERT_EQ(count_result, iterate_count); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestCreateCountReaderWithPredicateNotSupported) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_mor.db/pk_table_scan_and_read_mor/"; | |
| // Create splits from latest snapshot. | |
| ScanContextBuilder scan_context_builder(table_path); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| // Set predicate in read context. CountReader currently does not support predicate pushdown. | |
| auto predicate = PredicateBuilder::Equal(/*field_index=*/2, /*field_name=*/"f2", FieldType::INT, | |
| Literal(static_cast<int32_t>(0))); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| read_context_builder.SetPredicate(predicate); | |
| ASSERT_OK_AND_ASSIGN(auto read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| ASSERT_NOK_WITH_MSG(table_read->CreateCountReader(result_plan->Splits()), | |
| "predicate pushdown is not supported"); | |
| } | |
| TEST_P(ScanAndReadInteTest, TestCreateCountReaderWithForceKeepDeleteNotSupported) { | |
| auto file_format = FileFormat(); | |
| std::string table_path = paimon::test::GetDataDir() + file_format + | |
| "/pk_table_scan_and_read_mor.db/pk_table_scan_and_read_mor/"; | |
| // Create splits from latest snapshot. | |
| ScanContextBuilder scan_context_builder(table_path); | |
| ASSERT_OK_AND_ASSIGN(auto scan_context, FinishScanContext(scan_context_builder)); | |
| ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); | |
| ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); | |
| ReadContextBuilder read_context_builder(table_path); | |
| AddReadOptionsForPrefetch(&read_context_builder); | |
| ASSERT_OK_AND_ASSIGN(auto read_context, read_context_builder.Finish()); | |
| ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); | |
| auto* key_value_table_read = dynamic_cast<KeyValueTableRead*>(table_read.get()); | |
| ASSERT_TRUE(key_value_table_read != nullptr); | |
| key_value_table_read->ForceKeepDelete(true); | |
| ASSERT_NOK_WITH_MSG(table_read->CreateCountReader(result_plan->Splits()), | |
| "force_keep_delete is not supported"); | |
| } | |
| } // namespace paimon::test |