wangyong9999 commented on code in PR #284:
URL: https://github.com/apache/paimon-cpp/pull/284#discussion_r3953794153
##########
include/paimon/cache/cache.h:
##########
@@ -48,6 +48,16 @@ class PAIMON_EXPORT CacheKey {
const
std::string& branch,
int32_t
bucket);
+ /// Cache candidates for an inferred bucket, including files with other
bucket counts or
+ /// schemas.
+ /// @param total_buckets Positive bucket count used to compute the
inferred bucket.
+ /// @param schema_id Schema used to build the selector.
+ static std::shared_ptr<CacheKey> ForSnapshotLiveManifestEntries(const
std::string& table_path,
+ const
std::string& branch,
+ int32_t
bucket,
+ int32_t
total_buckets,
+ int64_t
schema_id);
+
Review Comment:
Moved the new five-argument factory to the internal cache header in e6bb0bfb
and named it CreateInferredSnapshotLiveManifestEntriesCacheKey. Kept the
existing three-argument public factory for compatibility. The PR no longer adds
a public overload; cache isolation and Append/KV integration tests pass.
##########
src/paimon/common/io/cache/cache_key.cpp:
##########
@@ -24,11 +24,14 @@ namespace {
class SnapshotLiveManifestEntriesCacheKey : public CacheKey {
public:
SnapshotLiveManifestEntriesCacheKey(const std::string& table_path, const
std::string& branch,
- int32_t bucket)
+ int32_t bucket, int32_t total_buckets
= 0,
+ int64_t schema_id = 0)
Review Comment:
Removed the constructor default arguments in e6bb0bfb. The explicit-bucket
factory now passes both values explicitly.
##########
src/paimon/core/operation/key_value_file_store_scan_test.cpp:
##########
@@ -58,6 +60,51 @@ class Schema;
} // namespace arrow
namespace paimon::test {
+TEST(KeyValueBucketPruningTest, UsesEachEntriesBucketCount) {
+ auto pool = GetDefaultPool();
+ auto arrow_schema = DataField::ConvertDataFieldsToArrowSchema(
+ {DataField(0, arrow::field("rowkey", arrow::utf8()))});
+ std::map<std::string, std::string> options = {{Options::BUCKET, "4"}};
+ auto test_dir = UniqueTestDirectory::Create("local");
+ auto manager =
+ std::make_shared<SchemaManager>(std::make_shared<LocalFileSystem>(),
test_dir->Str());
+ ASSERT_OK(manager->CreateTable(arrow_schema, {}, {"rowkey"}, options));
+ for (int64_t current_schema_id : {0, 1}) {
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<TableSchema> schema,
+ TableSchema::Create(current_schema_id, arrow_schema, {},
{"rowkey"}, options));
+ ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options));
+ auto predicate = PredicateBuilder::Equal(0, "rowkey",
FieldType::STRING,
+ Literal(FieldType::STRING,
"key", 3));
+ auto filters = std::make_shared<ScanFilter>(
+ predicate, std::vector<std::map<std::string, std::string>>(),
std::nullopt);
+ ASSERT_OK_AND_ASSIGN(
+ auto scan,
+ KeyValueFileStoreScan::Create(nullptr, manager, nullptr, nullptr,
schema, arrow_schema,
+ filters, core_options,
GetGlobalDefaultExecutor(), pool));
Review Comment:
Replaced the global executor with CreateDefaultExecutor() in e6bb0bfb. The
focused KV scan tests pass.
##########
src/paimon/core/operation/append_only_file_store_scan_test.cpp:
##########
@@ -46,10 +51,172 @@
#include "paimon/status.h"
#include "paimon/table/source/scan_metrics.h"
#include "paimon/table/source/table_scan.h"
+#include "paimon/testing/utils/binary_row_generator.h"
#include "paimon/testing/utils/testharness.h"
#include "paimon/testing/utils/timezone_guard.h"
namespace paimon::test {
+class AppendBucketPruningTest : public testing::Test {
+ public:
+ Result<std::unique_ptr<AppendOnlyFileStoreScan>> CreateScan(
+ const std::shared_ptr<Predicate>& predicate,
+ const std::optional<int32_t>& bucket = std::nullopt) const {
+ std::vector<DataField> fields = {DataField(0, arrow::field("rowkey",
rowkey_type_)),
+ DataField(1, arrow::field("value",
arrow::int32()))};
+ if (extra_field_) {
+ fields.emplace_back(2, arrow::field("extra", arrow::int32()));
+ }
+ auto arrow_schema = DataField::ConvertDataFieldsToArrowSchema(fields);
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<TableSchema> schema,
+ TableSchema::Create(schema_id_, arrow_schema,
{}, {}, options_));
+ PAIMON_ASSIGN_OR_RAISE(CoreOptions options,
CoreOptions::FromMap(options_));
+ auto filters = std::make_shared<ScanFilter>(
+ predicate, std::vector<std::map<std::string, std::string>>(),
bucket);
+ return AppendOnlyFileStoreScan::Create(nullptr, schema_manager_,
nullptr, nullptr, schema,
+ arrow_schema, filters, options,
+ GetGlobalDefaultExecutor(),
pool_);
Review Comment:
Replaced the global executor with CreateDefaultExecutor() in e6bb0bfb. The
focused Append scan tests pass.
##########
src/paimon/core/operation/append_only_file_store_scan_test.cpp:
##########
@@ -46,10 +51,172 @@
#include "paimon/status.h"
#include "paimon/table/source/scan_metrics.h"
#include "paimon/table/source/table_scan.h"
+#include "paimon/testing/utils/binary_row_generator.h"
#include "paimon/testing/utils/testharness.h"
#include "paimon/testing/utils/timezone_guard.h"
namespace paimon::test {
+class AppendBucketPruningTest : public testing::Test {
+ public:
+ Result<std::unique_ptr<AppendOnlyFileStoreScan>> CreateScan(
+ const std::shared_ptr<Predicate>& predicate,
+ const std::optional<int32_t>& bucket = std::nullopt) const {
+ std::vector<DataField> fields = {DataField(0, arrow::field("rowkey",
rowkey_type_)),
+ DataField(1, arrow::field("value",
arrow::int32()))};
+ if (extra_field_) {
+ fields.emplace_back(2, arrow::field("extra", arrow::int32()));
+ }
+ auto arrow_schema = DataField::ConvertDataFieldsToArrowSchema(fields);
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<TableSchema> schema,
+ TableSchema::Create(schema_id_, arrow_schema,
{}, {}, options_));
+ PAIMON_ASSIGN_OR_RAISE(CoreOptions options,
CoreOptions::FromMap(options_));
+ auto filters = std::make_shared<ScanFilter>(
+ predicate, std::vector<std::map<std::string, std::string>>(),
bucket);
+ return AppendOnlyFileStoreScan::Create(nullptr, schema_manager_,
nullptr, nullptr, schema,
+ arrow_schema, filters, options,
+ GetGlobalDefaultExecutor(),
pool_);
+ }
+
+ void CheckBuckets(const std::shared_ptr<Predicate>& predicate,
+ const std::optional<int32_t>& expected_bucket,
+ const std::optional<int32_t>& explicit_bucket =
std::nullopt,
+ int32_t total_buckets = kNumBuckets) const {
+ ASSERT_OK_AND_ASSIGN(auto scan, CreateScan(predicate,
explicit_bucket));
+ // Every file's stats match the lookup, so only bucket pruning can
discard a file.
+ SimpleStats stats = BinaryRowGenerator::GenerateStats(
+ {std::string("a"), 0}, {std::string("z"), 100}, {0, 0},
pool_.get());
+ ASSERT_OK_AND_ASSIGN(
+ auto file,
+ DataFileMeta::ForAppend("data.parquet", 100, 10, stats, 0, 9, 0,
std::nullopt,
+ std::nullopt, std::nullopt, std::nullopt,
std::nullopt));
+ for (int32_t bucket = 0; bucket < total_buckets; ++bucket) {
+ ManifestEntry entry(FileKind::Add(), BinaryRow::EmptyRow(),
bucket, total_buckets,
+ file);
+ ASSERT_OK_AND_ASSIGN(auto stats_scan, CreateScan(predicate,
bucket));
+ ASSERT_OK_AND_ASSIGN(bool stats_match,
stats_scan->FilterByStats(entry));
+ ASSERT_TRUE(stats_match);
+ ASSERT_OK_AND_ASSIGN(bool keep, scan->FilterManifestEntry(entry));
+ ASSERT_EQ(keep, !expected_bucket.has_value() || bucket ==
expected_bucket.value());
+ }
+ }
+
+ std::shared_ptr<Predicate> KeyEquals() const {
+ return PredicateBuilder::Equal(0, "rowkey", FieldType::STRING,
+ Literal(FieldType::STRING, "key", 3));
+ }
+
+ template <typename T>
+ void CheckMatchingValue(FieldType field_type, const T& query_value, const
T& stored_value) {
+ options_[Options::BUCKET] = "17";
+ auto predicate = PredicateBuilder::Equal(0, "rowkey", field_type,
Literal(query_value));
+ ASSERT_OK_AND_ASSIGN(auto comparison,
+
Literal(query_value).CompareTo(Literal(stored_value)));
+ ASSERT_EQ(comparison, 0);
+ BinaryRow stored_key = BinaryRowGenerator::GenerateRow({stored_value},
pool_.get());
+ BinaryRow query_key = BinaryRowGenerator::GenerateRow({query_value},
pool_.get());
+ int32_t bucket = DefaultBucketFunction().Bucket(stored_key, 17);
+ ASSERT_NE(bucket, DefaultBucketFunction().Bucket(query_key, 17));
+ SimpleStats stats = BinaryRowGenerator::GenerateStats(
+ {stored_value, 0}, {stored_value, 100}, {0, 0}, pool_.get());
+ ASSERT_OK_AND_ASSIGN(
+ auto file,
+ DataFileMeta::ForAppend("data.parquet", 100, 10, stats, 0, 9, 0,
std::nullopt,
+ std::nullopt, std::nullopt, std::nullopt,
std::nullopt));
+ ManifestEntry entry(FileKind::Add(), BinaryRow::EmptyRow(), bucket,
17, file);
+ ASSERT_OK_AND_ASSIGN(auto stats_scan, CreateScan(predicate, bucket));
+ ASSERT_OK_AND_ASSIGN(bool stats_match,
stats_scan->FilterByStats(entry));
+ ASSERT_TRUE(stats_match);
+ ASSERT_OK_AND_ASSIGN(auto scan, CreateScan(predicate));
+ ASSERT_OK_AND_ASSIGN(bool keep, scan->FilterManifestEntry(entry));
+ ASSERT_TRUE(keep);
+ }
+
+ std::shared_ptr<arrow::DataType> rowkey_type_ = arrow::utf8();
+ static constexpr int32_t kNumBuckets = 4;
+ int64_t schema_id_ = 0;
+ bool extra_field_ = false;
+ std::shared_ptr<SchemaManager> schema_manager_;
+ std::shared_ptr<MemoryPool> pool_ = GetDefaultPool();
+ std::map<std::string, std::string> options_ = {{Options::BUCKET, "4"},
+ {Options::BUCKET_KEY,
"rowkey"}};
+};
+
+TEST_F(AppendBucketPruningTest, PrunesStringKeyLookup) {
+ BinaryRow key = BinaryRowGenerator::GenerateRow({std::string("key")},
pool_.get());
+ int32_t expected_bucket = DefaultBucketFunction().Bucket(key, kNumBuckets);
+ CheckBuckets(KeyEquals(), expected_bucket);
+}
+
+TEST_F(AppendBucketPruningTest, PreservesExplicitBucketFilter) {
+ BinaryRow key = BinaryRowGenerator::GenerateRow({std::string("key")},
pool_.get());
+ int32_t other_bucket = (DefaultBucketFunction().Bucket(key, kNumBuckets) +
1) % kNumBuckets;
+ CheckBuckets(KeyEquals(), other_bucket, other_bucket);
+}
+
+TEST_F(AppendBucketPruningTest, DoesNotPruneWithoutCompleteEqualKeys) {
+ CheckBuckets(nullptr, std::nullopt);
+ CheckBuckets(PredicateBuilder::GreaterThan(0, "rowkey", FieldType::STRING,
+ Literal(FieldType::STRING,
"key", 3)),
+ std::nullopt);
+ options_[Options::BUCKET_KEY] = "rowkey,value";
+ CheckBuckets(KeyEquals(), std::nullopt);
+}
+
+TEST_F(AppendBucketPruningTest, DoesNotPruneBucketUnawareTable) {
+ options_[Options::BUCKET] = "-1";
+ CheckBuckets(KeyEquals(), std::nullopt);
+}
+
+TEST_F(AppendBucketPruningTest, UsesEachEntriesBucketCount) {
+ BinaryRow key = BinaryRowGenerator::GenerateRow({std::string("key")},
pool_.get());
+ for (int32_t total_buckets : {2, 4, 8, 17}) {
+ CheckBuckets(KeyEquals(), DefaultBucketFunction().Bucket(key,
total_buckets), std::nullopt,
+ total_buckets);
+ }
+}
+
+TEST_F(AppendBucketPruningTest, PrunesCompatibleHistoricalSchema) {
+ auto test_dir = UniqueTestDirectory::Create("local");
+ schema_manager_ =
+ std::make_shared<SchemaManager>(std::make_shared<LocalFileSystem>(),
test_dir->Str());
+ ASSERT_OK_AND_ASSIGN(auto scan, CreateScan(nullptr));
+ ASSERT_OK(schema_manager_->CreateTable(scan->schema_, {}, {}, options_));
+ schema_id_ = 1;
+ extra_field_ = true;
+ BinaryRow key = BinaryRowGenerator::GenerateRow({std::string("key")},
pool_.get());
+ CheckBuckets(KeyEquals(), DefaultBucketFunction().Bucket(key,
kNumBuckets));
+}
+
+TEST_F(AppendBucketPruningTest, PreservesIncompatibleHistoricalBucketKeys) {
+ auto test_dir = UniqueTestDirectory::Create("local");
+ schema_manager_ =
+ std::make_shared<SchemaManager>(std::make_shared<LocalFileSystem>(),
test_dir->Str());
+ ASSERT_OK_AND_ASSIGN(auto scan, CreateScan(nullptr));
+ ASSERT_OK(schema_manager_->CreateTable(scan->schema_, {}, {}, options_));
+ schema_id_ = 1;
+ rowkey_type_ = arrow::binary();
+ auto predicate = PredicateBuilder::Equal(0, "rowkey", FieldType::BINARY,
+ Literal(FieldType::BINARY, "key",
3));
+ CheckBuckets(predicate, std::nullopt);
+}
+
+TEST_F(AppendBucketPruningTest, PreservesCrossScaleDecimalMatch) {
+ rowkey_type_ = arrow::decimal128(10, 2);
+ CheckMatchingValue(FieldType::DECIMAL, Decimal::FromUnscaledLong(12, 10,
1),
+ Decimal::FromUnscaledLong(120, 10, 2));
+}
+
+TEST_F(AppendBucketPruningTest, PreservesDifferentNaNPayloadMatch) {
+ rowkey_type_ = arrow::float64();
+ uint64_t query_bits = 0x7ff8000000000000ULL;
+ uint64_t stored_bits = 0x7ff8000000000001ULL;
+ double query_value;
+ double stored_value;
+ std::memcpy(&query_value, &query_bits, sizeof(query_value));
+ std::memcpy(&stored_value, &stored_bits, sizeof(stored_value));
+ CheckMatchingValue(FieldType::DOUBLE, query_value, stored_value);
Review Comment:
Replaced memcpy with FloatingPointFromBits from math.h in e6bb0bfb.
CanonicalizeFloatingPoint would collapse the two NaN payloads and remove the
regression this test covers. The test still checks that a matching file with a
different stored NaN payload is retained, and passes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
