From fc257729b29529d853097c153e9d0337c1ec9b23 Mon Sep 17 00:00:00 2001 From: "lisizhuo.lsz" Date: Wed, 4 Nov 2026 17:35:05 +0800 Subject: [PATCH] test: cover global indexes, nested projections and changelog scans --- .../key_value_file_store_scan_test.cpp | 40 ++++ .../utils/nested_projection_utils_test.cpp | 64 ++++++ test/inte/global_index_test.cpp | 209 ++++++++++++++++++ 3 files changed, 313 insertions(+) diff --git a/src/paimon/core/operation/key_value_file_store_scan_test.cpp b/src/paimon/core/operation/key_value_file_store_scan_test.cpp index 253d3c3d5..963d75274 100644 --- a/src/paimon/core/operation/key_value_file_store_scan_test.cpp +++ b/src/paimon/core/operation/key_value_file_store_scan_test.cpp @@ -38,6 +38,7 @@ #include "paimon/core/schema/table_schema.h" #include "paimon/core/snapshot.h" #include "paimon/core/stats/simple_stats.h" +#include "paimon/core/table/source/scan_mode.h" #include "paimon/core/utils/field_mapping.h" #include "paimon/core/utils/file_store_path_factory.h" #include "paimon/core/utils/snapshot_manager.h" @@ -498,6 +499,45 @@ TEST_F(KeyValueFileStoreScanTest, TestFilterByValueFilterWithValueStatsCols) { ASSERT_FALSE(keep); } +TEST_F(KeyValueFileStoreScanTest, TestChangelogValueFilterByProducer) { + auto schema = DataField::ConvertDataFieldsToArrowSchema( + {DataField(0, arrow::field("id", arrow::int32())), + DataField(1, arrow::field("value", arrow::int32()))}); + auto predicate = PredicateBuilder::GreaterThan(1, "value", FieldType::INT, Literal(30)); + auto scan_filter = std::make_shared( + predicate, std::vector>(), std::nullopt); + std::shared_ptr executor = CreateDefaultExecutor(); + std::vector> producers = { + {"lookup", true}, {"full-compaction", true}, {"input", false}}; + for (const auto& [producer, value_filter_enabled] : producers) { + std::map options = {{Options::CHANGELOG_PRODUCER, producer}}; + ASSERT_OK_AND_ASSIGN(CoreOptions core_options, CoreOptions::FromMap(options)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr table_schema, + TableSchema::Create(0, schema, {}, {"id"}, options)); + ASSERT_OK_AND_ASSIGN(std::unique_ptr scan, + KeyValueFileStoreScan::Create( + /*snapshot_manager=*/nullptr, /*schema_manager=*/nullptr, + /*manifest_list=*/nullptr, /*manifest_file=*/nullptr, table_schema, + schema, scan_filter, core_options, executor, pool_)); + scan->WithKind(ScanMode::CHANGELOG); + + // Lookup/full-compaction prune nonmatching files without EnableValueFilter(). + // Input changelogs must retain them; matching files are kept for every producer. + for (int32_t value : {20, 40}) { + SimpleStats stats = BinaryRowGenerator::GenerateStats( + /*min=*/{1, value}, /*max=*/{1, value}, /*null=*/{0, 0}, pool_.get()); + ASSERT_OK_AND_ASSIGN( + std::shared_ptr file, + DataFileMeta::ForAppend("changelog.parquet", 100, 1, stats, 0, 0, 0, std::nullopt, + std::nullopt, std::nullopt, std::nullopt, std::nullopt)); + ManifestEntry entry(FileKind::Add(), BinaryRow::EmptyRow(), /*bucket=*/0, + /*total_buckets=*/1, file); + ASSERT_OK_AND_ASSIGN(bool keep, scan->FilterByStats(entry)); + ASSERT_EQ(keep, !value_filter_enabled || value > 30); + } + } +} + TEST_F(KeyValueFileStoreScanTest, TestFilterByValueFilterWithSchemaEvolution) { std::string table_path = paimon::test::GetDataDir() + "orc/pk_table_with_alter_table.db/pk_table_with_alter_table"; diff --git a/src/paimon/core/utils/nested_projection_utils_test.cpp b/src/paimon/core/utils/nested_projection_utils_test.cpp index 9ca69e91e..3d33910be 100644 --- a/src/paimon/core/utils/nested_projection_utils_test.cpp +++ b/src/paimon/core/utils/nested_projection_utils_test.cpp @@ -313,6 +313,31 @@ TEST(NestedProjectionUtilsTest, PruneDataTypeMapStructSchemaEvolutionAddedField) ASSERT_TRUE(result.value()->Equals(*data_type)) << result.value()->ToString(); } +TEST(NestedProjectionUtilsTest, PruneDataTypeNestedListMapWithAddedFieldAndVariantAccess) { + auto wrap_in_containers = [](const std::shared_ptr& value_type) { + return arrow::list( + arrow::list(arrow::map(arrow::utf8(), value_type, /*keys_sorted=*/true))); + }; + auto data_value = + arrow::struct_({MakeField("a", arrow::int32(), 10), + MakeField("v", VariantTypeUtils::UnshreddedStructType(), 11)}); + auto read_value = arrow::struct_({MakeField("a", arrow::int32(), 10), + MakeField("v", MakeVariantAccessType(), 11), + MakeField("added", arrow::utf8(), 12)}); + // Only fields present in the file are retained, with the variant projection applied. + auto expected_value = arrow::struct_( + {MakeField("a", arrow::int32(), 10), MakeField("v", MakeVariantAccessType(), 11)}); + auto data_type = wrap_in_containers(data_value); + auto read_type = wrap_in_containers(read_value); + auto expected_type = wrap_in_containers(expected_value); + + ASSERT_OK_AND_ASSIGN(std::optional> pruned, + NestedProjectionUtils::PruneDataType(read_type, data_type)); + ASSERT_TRUE(pruned.has_value()); + ASSERT_TRUE(pruned.value()->Equals(*expected_type, /*check_metadata=*/true)) + << pruned.value()->ToString(); +} + TEST(NestedProjectionUtilsTest, PruneDataTypeListStructDropAndAddStillFails) { // Dropping a file field (b) is a real partial projection -- keep failing. auto data_inner = @@ -641,6 +666,45 @@ TEST(NestedProjectionUtilsTest, HasNestedSubfieldProjectionWithProjection) { ASSERT_TRUE(has_nested_projection); } +TEST(NestedProjectionUtilsTest, HasNestedSubfieldProjectionVariantAccessReturnsFalse) { + auto file_schema = arrow::schema({MakeField("v", VariantTypeUtils::UnshreddedStructType(), 1)}); + auto read_schema = arrow::schema({MakeField("v", MakeVariantAccessType(), 1)}); + + ASSERT_OK_AND_ASSIGN( + bool has_nested_projection, + NestedProjectionUtils::HasNestedSubfieldProjection(file_schema, read_schema)); + ASSERT_FALSE(has_nested_projection); +} + +TEST(NestedProjectionUtilsTest, HasNestedSubfieldProjectionNestedStruct) { + auto file_child = + arrow::struct_({MakeField("a", arrow::int32(), 3), MakeField("b", arrow::utf8(), 4)}); + auto read_child = arrow::struct_({MakeField("a", arrow::int32(), 3)}); + // The outer struct keeps all its fields; only the child struct is projected. + auto file_schema = + arrow::schema({MakeField("f0", arrow::struct_({MakeField("child", file_child, 2)}), 1)}); + auto read_schema = + arrow::schema({MakeField("f0", arrow::struct_({MakeField("child", read_child, 2)}), 1)}); + + ASSERT_OK_AND_ASSIGN( + bool has_nested_projection, + NestedProjectionUtils::HasNestedSubfieldProjection(file_schema, read_schema)); + ASSERT_TRUE(has_nested_projection); +} + +TEST(NestedProjectionUtilsTest, HasNestedSubfieldProjectionMapKey) { + auto file_key = + arrow::struct_({MakeField("a", arrow::int32(), 2), MakeField("b", arrow::utf8(), 3)}); + auto read_key = arrow::struct_({MakeField("a", arrow::int32(), 2)}); + auto file_schema = arrow::schema({MakeField("f0", arrow::map(file_key, arrow::int64()), 1)}); + auto read_schema = arrow::schema({MakeField("f0", arrow::map(read_key, arrow::int64()), 1)}); + + ASSERT_OK_AND_ASSIGN( + bool has_nested_projection, + NestedProjectionUtils::HasNestedSubfieldProjection(file_schema, read_schema)); + ASSERT_TRUE(has_nested_projection); +} + TEST(NestedProjectionUtilsTest, HasNestedSubfieldProjectionTypeMismatchReturnsInvalid) { auto file_schema = arrow::schema({ MakeField("f0", arrow::struct_({MakeField("a", arrow::int32(), 2)}), 1), diff --git a/test/inte/global_index_test.cpp b/test/inte/global_index_test.cpp index 19c606ef0..b92b44bf6 100644 --- a/test/inte/global_index_test.cpp +++ b/test/inte/global_index_test.cpp @@ -20,6 +20,7 @@ #include #include +#include "arrow/c/bridge.h" #include "arrow/type.h" #include "fmt/format.h" #include "fmt/ranges.h" @@ -47,6 +48,8 @@ #include "paimon/global_index/global_index_result.h" #include "paimon/global_index/global_index_scan.h" #include "paimon/global_index/global_index_write_task.h" +#include "paimon/global_index/global_indexer.h" +#include "paimon/global_index/global_indexer_factory.h" #include "paimon/predicate/literal.h" #include "paimon/predicate/predicate_builder.h" #include "paimon/result.h" @@ -103,6 +106,66 @@ class DeleteAfterListFileSystem final : public LocalFileSystem { mutable std::atomic deleted_ = false; }; +#ifdef PAIMON_ENABLE_ORC +class NestedFieldsIndexWriter : public GlobalIndexWriter { + public: + explicit NestedFieldsIndexWriter(std::shared_ptr type) + : type_(std::move(type)) {} + + Status AddBatch(::ArrowArray* array, std::vector&&) override { + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr decoded, + arrow::ImportArray(array, type_)); + return ToPaimonStatus(decoded->ValidateFull()); + } + + Result> Finish() override { + return std::vector{}; + } + + private: + std::shared_ptr type_; +}; + +class NestedFieldsIndexer : public GlobalIndexer { + public: + Result>> GetExtraFieldNames() const override { + return std::optional>( + {"details", "attributes", "plain_details", "plain_attributes"}); + } + + Result> CreateWriter( + const std::string& field_name, ::ArrowSchema* arrow_schema, + const std::shared_ptr& file_writer, + const std::shared_ptr& pool) const override { + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr type, + arrow::ImportType(arrow_schema)); + return std::shared_ptr( + std::make_shared(std::move(type))); + } + + Result> CreateReader( + ::ArrowSchema* arrow_schema, const std::shared_ptr& file_reader, + const std::vector& files, int64_t total_row_count, + const std::optional>& row_ranges, + const std::shared_ptr& pool) const override { + return Status::NotImplemented("test index has no reader"); + } +}; + +class NestedFieldsIndexFactory : public GlobalIndexerFactory { + public: + const char* Identifier() const override { + return "nested-fields-test-global"; + } + + Result> Create( + const std::map& options) const override { + return std::unique_ptr(std::make_unique()); + } +}; +REGISTER_PAIMON_FACTORY(NestedFieldsIndexFactory); +#endif + } // namespace using ScoredRows = std::vector>; @@ -1121,6 +1184,78 @@ TEST_P(GlobalIndexTest, TestWriteIndex) { } } +#ifdef PAIMON_ENABLE_ORC +TEST_P(GlobalIndexTest, TestWriteIndexDecodesNestedOrcDictionaries) { + if (file_format_ != "orc") { + GTEST_SKIP() << "ORC-only dictionary encoding case"; + } + arrow::FieldVector fields = { + arrow::field("key", arrow::int32()), + arrow::field("details", arrow::struct_({arrow::field("text", arrow::utf8()), + arrow::field("number", arrow::int32())})), + arrow::field("attributes", arrow::map(arrow::utf8(), arrow::utf8())), + arrow::field("plain_details", arrow::struct_({arrow::field("left", arrow::int32()), + arrow::field("right", arrow::int32())})), + arrow::field("plain_attributes", arrow::map(arrow::int32(), arrow::int32()))}; + std::shared_ptr schema = arrow::schema(fields); + std::map options = {{Options::FILE_FORMAT, "orc"}, + {Options::FILE_SYSTEM, "local"}, + {Options::ROW_TRACKING_ENABLED, "true"}, + {Options::DATA_EVOLUTION_ENABLED, "true"}, + {"orc.dictionary-key-size-threshold", "1"}, + {"orc.read.enable-lazy-decoding", "true"}}; + CreateTable(/*partition_keys=*/{}, schema, options); + std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar"); + std::shared_ptr source = + arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([ + [0, ["red", 10], [["color", "red"]], [1, 2], [[1, 2]]], + [1, ["blue", 20], [["color", "blue"]], [3, 4], [[3, 4]]], + [2, ["red", 30], [["color", "red"]], [5, 6], [[5, 6]]], + [3, ["blue", 40], [["color", "blue"]], [7, 8], [[7, 8]]] + ])") + .ValueOrDie(); + ASSERT_OK_AND_ASSIGN(std::vector> messages, + WriteArray(table_path, schema->field_names(), source)); + ASSERT_OK(Commit(table_path, messages)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr split, + ScanData(table_path, /*partition_filters=*/{})); + std::shared_ptr indexed_split = + std::make_shared(split, std::vector{Range(0, 3)}); + ReadContextBuilder read_builder(table_path); + read_builder.SetOptions(options).WithFileSystem(fs_).SetReadFieldNames( + {"key", "details", "attributes", "plain_details", "plain_attributes", + SpecialFields::RowId().Name()}); + ASSERT_OK_AND_ASSIGN(std::unique_ptr read_context, read_builder.Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr table_read, + TableRead::Create(std::move(read_context))); + ASSERT_OK_AND_ASSIGN(std::unique_ptr reader, + table_read->CreateReader(indexed_split)); + ASSERT_OK_AND_ASSIGN(BatchReader::ReadBatch batch, reader->NextBatch()); + ASSERT_FALSE(BatchReader::IsEofBatch(batch)); + arrow::Result> raw_result = + arrow::ImportArray(batch.first.get(), batch.second.get()); + ASSERT_TRUE(raw_result.ok()) << raw_result.status().ToString(); + std::shared_ptr raw = std::move(raw_result).ValueOrDie(); + std::shared_ptr raw_rows = + std::dynamic_pointer_cast(raw); + ASSERT_TRUE(raw_rows); + std::shared_ptr raw_details = + std::dynamic_pointer_cast(raw_rows->GetFieldByName("details")); + std::shared_ptr raw_attributes = + std::dynamic_pointer_cast(raw_rows->GetFieldByName("attributes")); + ASSERT_TRUE(raw_details); + ASSERT_TRUE(raw_attributes); + std::shared_ptr raw_text = raw_details->GetFieldByName("text"); + ASSERT_TRUE(raw_text); + ASSERT_EQ(raw_text->type_id(), arrow::Type::DICTIONARY); + ASSERT_EQ(raw_attributes->keys()->type_id(), arrow::Type::DICTIONARY); + ASSERT_EQ(raw_attributes->items()->type_id(), arrow::Type::DICTIONARY); + ASSERT_OK( + GlobalIndexWriteTask::WriteIndex(table_path, "key", "nested-fields-test", indexed_split, + /*options=*/{}, /*task_id=*/std::nullopt, pool_, fs_)); +} +#endif + TEST_P(GlobalIndexTest, TestWriteIndexWithPartition) { CreateTable(/*partition_keys=*/{"f1"}); std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar"); @@ -1283,6 +1418,46 @@ TEST_P(GlobalIndexTest, TestScanIndex) { ASSERT_OK_AND_ASSIGN(auto index_result, global_index_scan_impl->Scan(predicate)); ASSERT_EQ(index_result->ToString(), "{2,3,6}"); } + { + // test starts with predicate for f0 + ASSERT_OK_AND_ASSIGN( + std::shared_ptr predicate, + PredicateBuilder::StartsWith(/*field_index=*/0, /*field_name=*/"f0", FieldType::STRING, + Literal(FieldType::STRING, "Al", 2))); + ASSERT_OK_AND_ASSIGN(auto index_result, global_index_scan_impl->Scan(predicate)); + ASSERT_TRUE(index_result); + ASSERT_EQ(index_result->ToString(), "{0,7}"); + } + { + // test ends with predicate for f0 + ASSERT_OK_AND_ASSIGN( + std::shared_ptr predicate, + PredicateBuilder::EndsWith(/*field_index=*/0, /*field_name=*/"f0", FieldType::STRING, + Literal(FieldType::STRING, "ce", 2))); + ASSERT_OK_AND_ASSIGN(auto index_result, global_index_scan_impl->Scan(predicate)); + ASSERT_TRUE(index_result); + ASSERT_EQ(index_result->ToString(), "{0,7}"); + } + { + // test contains predicate for f0 + ASSERT_OK_AND_ASSIGN( + std::shared_ptr predicate, + PredicateBuilder::Contains(/*field_index=*/0, /*field_name=*/"f0", FieldType::STRING, + Literal(FieldType::STRING, "lic", 3))); + ASSERT_OK_AND_ASSIGN(auto index_result, global_index_scan_impl->Scan(predicate)); + ASSERT_TRUE(index_result); + ASSERT_EQ(index_result->ToString(), "{0,7}"); + } + { + // test like predicate for f0 + ASSERT_OK_AND_ASSIGN( + std::shared_ptr predicate, + PredicateBuilder::Like(/*field_index=*/0, /*field_name=*/"f0", FieldType::STRING, + Literal(FieldType::STRING, "A_i%", 4))); + ASSERT_OK_AND_ASSIGN(auto index_result, global_index_scan_impl->Scan(predicate)); + ASSERT_TRUE(index_result); + ASSERT_EQ(index_result->ToString(), "{0,7}"); + } { // test and predicate auto f0_predicate = @@ -1701,6 +1876,40 @@ TEST_P(GlobalIndexTest, TestScanIndexWithPartition) { check_result(std::nullopt); } +TEST_P(GlobalIndexTest, TestScanIndexWithPartitionPredicate) { + if (file_format_ == "avro") { + return; + } + + // Only f1=10 has an index in this table. + std::string table_path = + paimon::test::GetDataDir() + "/" + file_format_ + + "/append_with_global_index_with_partition.db/append_with_global_index_with_partition"; + auto check_result = [&](const std::shared_ptr& partitions, + const std::shared_ptr& pool, bool expect_index) { + ASSERT_OK_AND_ASSIGN( + std::unique_ptr scan, + GlobalIndexScan::Create(table_path, /*snapshot_id=*/std::nullopt, partitions, + /*options=*/{}, fs_, /*executor=*/nullptr, pool)); + ASSERT_OK_AND_ASSIGN(auto readers, scan->CreateReaders("f0", std::nullopt)); + if (!expect_index) { + ASSERT_TRUE(readers.empty()); + return; + } + ASSERT_EQ(readers.size(), 1u); + ASSERT_OK_AND_ASSIGN(auto result, + readers[0]->VisitEqual(Literal(FieldType::STRING, "Bob", 3))); + ASSERT_TRUE(result); + ASSERT_EQ(result->ToString(), "{1,4}"); + }; + + check_result(/*partitions=*/nullptr, /*pool=*/nullptr, /*expect_index=*/true); + check_result(PredicateBuilder::Equal(/*field_index=*/0, "f1", FieldType::INT, Literal(10)), + pool_, /*expect_index=*/true); + check_result(PredicateBuilder::Equal(/*field_index=*/0, "f1", FieldType::INT, Literal(20)), + pool_, /*expect_index=*/false); +} + TEST_P(GlobalIndexTest, TestScanUnregisteredIndex) { if (file_format_ == "avro") { return;