This is an automated email from the ASF dual-hosted git repository.
lxy-9602 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new d46b25c1 fix(scan): share bucket pruning across append and key-value
scans (#284)
d46b25c1 is described below
commit d46b25c1ed28bb776685b33468a1054d58f986f2
Author: wangyong9999 <[email protected]>
AuthorDate: Wed Sep 9 11:47:28 2026 +0800
fix(scan): share bucket pruning across append and key-value scans (#284)
---
docs/source/api/scan.rst | 26 +++
include/paimon/cache/cache.h | 3 -
src/paimon/common/io/cache/cache_key.cpp | 58 ++----
src/paimon/common/io/cache/cache_key.h | 65 ++++++
src/paimon/common/io/cache/lru_cache_test.cpp | 34 +++-
src/paimon/core/bucket/bucket_select_converter.cpp | 45 ++++-
src/paimon/core/bucket/bucket_select_converter.h | 27 +++
.../core/bucket/bucket_select_converter_test.cpp | 74 +++++++
.../core/operation/append_only_file_store_scan.cpp | 2 +
.../core/operation/append_only_file_store_scan.h | 1 +
.../operation/append_only_file_store_scan_test.cpp | 163 ++++++++++++++++
src/paimon/core/operation/file_store_scan.cpp | 82 ++++++--
src/paimon/core/operation/file_store_scan.h | 15 +-
.../core/operation/key_value_file_store_scan.cpp | 21 --
.../operation/key_value_file_store_scan_test.cpp | 47 +++++
src/paimon/core/schema/schema_manager.cpp | 8 +-
src/paimon/core/schema/schema_manager.h | 3 +-
src/paimon/core/schema/schema_manager_test.cpp | 19 +-
test/inte/scan_and_read_inte_test.cpp | 217 ++++++++++++++++++---
19 files changed, 768 insertions(+), 142 deletions(-)
diff --git a/docs/source/api/scan.rst b/docs/source/api/scan.rst
index 142a5168..0194aa6d 100644
--- a/docs/source/api/scan.rst
+++ b/docs/source/api/scan.rst
@@ -21,6 +21,32 @@ Scan
.. _cpp-api-scan:
+Bucket pruning
+==============
+
+For fixed-bucket append and primary-key tables, an equality predicate on every
bucket key lets
+the scan derive the target bucket using the table's bucket function. Other
buckets
+are excluded from the scan plan without requiring an explicit bucket ID from
the
+caller. An explicit bucket filter takes precedence. Queries that do not
constrain
+all bucket keys with equality, and bucket-unaware tables, keep the existing
scan
+behavior. Both scan types use a shared selector that computes the bucket with
+each manifest entry's total bucket count, so rescaled files are not filtered
using
+the current table's bucket count. Historical schemas remain eligible when
their ordered bucket-key field IDs,
+types and bucket function match. Changes to unrelated columns do not disable
+pruning. Incompatible bucket schemas and nonpositive total bucket counts retain
+the existing filtering behavior.
+
+This inference prunes data files at the manifest-entry level. Manifest
min/max-bucket
+skipping requires an explicit bucket filter. When the snapshot
live-manifest-entry
+cache is enabled, inferred scans cache candidates by bucket, current bucket
count
+and current schema ID. Files with other bucket counts or schema IDs remain in
the
+cached candidates and are filtered after lookup. This permits cache reuse
without
+discarding files that require a different bucket calculation or schema
fallback.
+
+Decimal literals are rescaled to the bucket field's type only when the
conversion
+is exact. NaN literals and decimals that cannot be represented exactly disable
+inferred bucket pruning.
+
Interface
=========
diff --git a/include/paimon/cache/cache.h b/include/paimon/cache/cache.h
index 5bff2a12..b282da74 100644
--- a/include/paimon/cache/cache.h
+++ b/include/paimon/cache/cache.h
@@ -44,9 +44,6 @@ class PAIMON_EXPORT CacheKey {
int32_t length, bool
is_index);
static std::shared_ptr<CacheKey> ForKind(const std::string& file_path,
int64_t position,
int32_t length, CacheKind kind);
- static std::shared_ptr<CacheKey> ForSnapshotLiveManifestEntries(const
std::string& table_path,
- const
std::string& branch,
- int32_t
bucket);
public:
virtual ~CacheKey() = default;
diff --git a/src/paimon/common/io/cache/cache_key.cpp
b/src/paimon/common/io/cache/cache_key.cpp
index c84c8f90..998d387f 100644
--- a/src/paimon/common/io/cache/cache_key.cpp
+++ b/src/paimon/common/io/cache/cache_key.cpp
@@ -19,49 +19,6 @@
#include "paimon/common/io/cache/cache_key.h"
namespace paimon {
-namespace {
-
-class SnapshotLiveManifestEntriesCacheKey : public CacheKey {
- public:
- SnapshotLiveManifestEntriesCacheKey(const std::string& table_path, const
std::string& branch,
- int32_t bucket)
- : CacheKey(CacheKind::SNAPSHOT_LIVE_MANIFEST),
- table_path_(table_path),
- branch_(branch),
- bucket_(bucket) {}
-
- bool IsIndex() const override {
- return false;
- }
-
- bool Equals(const CacheKey& other) const override {
- const auto* rhs = dynamic_cast<const
SnapshotLiveManifestEntriesCacheKey*>(&other);
- if (!rhs) {
- return false;
- }
- return table_path_ == rhs->table_path_ && branch_ == rhs->branch_ &&
- bucket_ == rhs->bucket_ && GetKind() == rhs->GetKind();
- }
-
- size_t HashCode() const override {
- size_t seed = 0;
- seed ^= std::hash<std::string>{}(table_path_) + HASH_CONSTANT + (seed
<< 6) + (seed >> 2);
- seed ^= std::hash<std::string>{}(branch_) + HASH_CONSTANT + (seed <<
6) + (seed >> 2);
- seed ^= std::hash<int32_t>{}(bucket_) + HASH_CONSTANT + (seed << 6) +
(seed >> 2);
- seed ^= std::hash<int32_t>{}(static_cast<int32_t>(GetKind())) +
HASH_CONSTANT +
- (seed << 6) + (seed >> 2);
- return seed;
- }
-
- private:
- static constexpr uint64_t HASH_CONSTANT = 0x9e3779b97f4a7c15ULL;
-
- const std::string table_path_;
- const std::string branch_;
- const int32_t bucket_;
-};
-
-} // namespace
std::shared_ptr<CacheKey> CacheKey::ForPosition(const std::string& file_path,
int64_t position,
int32_t length, bool is_index)
{
@@ -76,10 +33,17 @@ std::shared_ptr<CacheKey> CacheKey::ForKind(const
std::string& file_path, int64_
return key;
}
-std::shared_ptr<CacheKey> CacheKey::ForSnapshotLiveManifestEntries(const
std::string& table_path,
- const
std::string& branch,
- int32_t
bucket) {
- return std::make_shared<SnapshotLiveManifestEntriesCacheKey>(table_path,
branch, bucket);
+std::shared_ptr<CacheKey> SnapshotLiveManifestEntriesCacheKey::ForExplicit(
+ const std::string& table_path, const std::string& branch, int32_t bucket) {
+ return std::shared_ptr<CacheKey>(new SnapshotLiveManifestEntriesCacheKey(
+ table_path, branch, bucket, Mode::kExplicit, std::nullopt,
std::nullopt));
+}
+
+std::shared_ptr<CacheKey> SnapshotLiveManifestEntriesCacheKey::ForInferred(
+ const std::string& table_path, const std::string& branch, int32_t bucket,
int32_t total_buckets,
+ int64_t schema_id) {
+ return std::shared_ptr<CacheKey>(new SnapshotLiveManifestEntriesCacheKey(
+ table_path, branch, bucket, Mode::kInferred, total_buckets,
schema_id));
}
bool PositionCacheKey::IsIndex() const {
diff --git a/src/paimon/common/io/cache/cache_key.h
b/src/paimon/common/io/cache/cache_key.h
index 988735d1..82fa9752 100644
--- a/src/paimon/common/io/cache/cache_key.h
+++ b/src/paimon/common/io/cache/cache_key.h
@@ -20,12 +20,77 @@
#include <cstdint>
#include <memory>
+#include <optional>
#include <string>
#include "paimon/cache/cache.h"
namespace paimon {
+class SnapshotLiveManifestEntriesCacheKey : public CacheKey {
+ public:
+ static std::shared_ptr<CacheKey> ForExplicit(const std::string& table_path,
+ const std::string& branch,
int32_t bucket);
+ static std::shared_ptr<CacheKey> ForInferred(const std::string& table_path,
+ const std::string& branch,
int32_t bucket,
+ int32_t total_buckets,
int64_t schema_id);
+
+ bool IsIndex() const override {
+ return false;
+ }
+
+ bool Equals(const CacheKey& other) const override {
+ const auto* rhs = dynamic_cast<const
SnapshotLiveManifestEntriesCacheKey*>(&other);
+ if (!rhs) {
+ return false;
+ }
+ return table_path_ == rhs->table_path_ && branch_ == rhs->branch_ &&
+ bucket_ == rhs->bucket_ && mode_ == rhs->mode_ &&
+ total_buckets_ == rhs->total_buckets_ && schema_id_ ==
rhs->schema_id_ &&
+ GetKind() == rhs->GetKind();
+ }
+
+ size_t HashCode() const override {
+ size_t seed = 0;
+ seed ^= std::hash<std::string>{}(table_path_) + HASH_CONSTANT + (seed
<< 6) + (seed >> 2);
+ seed ^= std::hash<std::string>{}(branch_) + HASH_CONSTANT + (seed <<
6) + (seed >> 2);
+ seed ^= std::hash<int32_t>{}(bucket_) + HASH_CONSTANT + (seed << 6) +
(seed >> 2);
+ seed ^= std::hash<int32_t>{}(static_cast<int32_t>(GetKind())) +
HASH_CONSTANT +
+ (seed << 6) + (seed >> 2);
+ seed ^= std::hash<std::optional<int32_t>>{}(total_buckets_) +
HASH_CONSTANT + (seed << 6) +
+ (seed >> 2);
+ seed ^= std::hash<std::optional<int64_t>>{}(schema_id_) +
HASH_CONSTANT + (seed << 6) +
+ (seed >> 2);
+ seed ^= std::hash<int32_t>{}(static_cast<int32_t>(mode_)) +
HASH_CONSTANT + (seed << 6) +
+ (seed >> 2);
+ return seed;
+ }
+
+ private:
+ enum class Mode { kExplicit, kInferred };
+
+ SnapshotLiveManifestEntriesCacheKey(const std::string& table_path, const
std::string& branch,
+ int32_t bucket, Mode mode,
+ std::optional<int32_t> total_buckets,
+ std::optional<int64_t> schema_id)
+ : CacheKey(CacheKind::SNAPSHOT_LIVE_MANIFEST),
+ table_path_(table_path),
+ branch_(branch),
+ bucket_(bucket),
+ mode_(mode),
+ total_buckets_(total_buckets),
+ schema_id_(schema_id) {}
+
+ static constexpr uint64_t HASH_CONSTANT = 0x9e3779b97f4a7c15ULL;
+
+ const std::string table_path_;
+ const std::string branch_;
+ const int32_t bucket_;
+ const Mode mode_;
+ const std::optional<int32_t> total_buckets_;
+ const std::optional<int64_t> schema_id_;
+};
+
class PositionCacheKey : public CacheKey {
public:
PositionCacheKey(const std::string& file_path, int64_t position, int32_t
length, bool is_index,
diff --git a/src/paimon/common/io/cache/lru_cache_test.cpp
b/src/paimon/common/io/cache/lru_cache_test.cpp
index 1d644c70..29eb10fe 100644
--- a/src/paimon/common/io/cache/lru_cache_test.cpp
+++ b/src/paimon/common/io/cache/lru_cache_test.cpp
@@ -383,14 +383,32 @@ TEST_F(LruCacheTest, TestForKindSetsKeyKind) {
ASSERT_EQ(CacheKind::MANIFEST, put_key->GetKind());
}
-TEST_F(LruCacheTest, TestForSnapshotLiveManifestEntries) {
- auto main_key = CacheKey::ForSnapshotLiveManifestEntries("table_path",
"main", 0);
- auto same_key = CacheKey::ForSnapshotLiveManifestEntries("table_path",
"main", 0);
- auto branch_key = CacheKey::ForSnapshotLiveManifestEntries("table_path",
"dev", 0);
- auto table_key =
CacheKey::ForSnapshotLiveManifestEntries("other_table_path", "main", 0);
- auto bucket_key = CacheKey::ForSnapshotLiveManifestEntries("table_path",
"main", 1);
- auto hash_in_path_key =
CacheKey::ForSnapshotLiveManifestEntries("table#path", "main", 0);
- auto hash_in_branch_key =
CacheKey::ForSnapshotLiveManifestEntries("table", "path#main", 0);
+TEST_F(LruCacheTest, InferredManifestCacheKeysIncludeBucketCountAndSchema) {
+ auto key = SnapshotLiveManifestEntriesCacheKey::ForInferred("table",
"main", 1, 4, 0);
+ auto same = SnapshotLiveManifestEntriesCacheKey::ForInferred("table",
"main", 1, 4, 0);
+ ASSERT_TRUE(key->Equals(*same));
+ ASSERT_EQ(key->HashCode(), same->HashCode());
+ auto explicit_key =
SnapshotLiveManifestEntriesCacheKey::ForExplicit("table", "main", 1);
+ // Schema zero is real inferred metadata, not an explicit-mode placeholder.
+ ASSERT_FALSE(key->Equals(*explicit_key));
+ ASSERT_FALSE(explicit_key->Equals(*key));
+ ASSERT_FALSE(
+ key->Equals(*SnapshotLiveManifestEntriesCacheKey::ForInferred("table",
"main", 1, 8, 0)));
+ ASSERT_FALSE(
+ key->Equals(*SnapshotLiveManifestEntriesCacheKey::ForInferred("table",
"main", 1, 4, 1)));
+}
+
+TEST_F(LruCacheTest, TestExplicitSnapshotLiveManifestEntriesKeys) {
+ auto main_key =
SnapshotLiveManifestEntriesCacheKey::ForExplicit("table_path", "main", 0);
+ auto same_key =
SnapshotLiveManifestEntriesCacheKey::ForExplicit("table_path", "main", 0);
+ auto branch_key =
SnapshotLiveManifestEntriesCacheKey::ForExplicit("table_path", "dev", 0);
+ auto table_key =
+ SnapshotLiveManifestEntriesCacheKey::ForExplicit("other_table_path",
"main", 0);
+ auto bucket_key =
SnapshotLiveManifestEntriesCacheKey::ForExplicit("table_path", "main", 1);
+ auto hash_in_path_key =
+ SnapshotLiveManifestEntriesCacheKey::ForExplicit("table#path", "main",
0);
+ auto hash_in_branch_key =
+ SnapshotLiveManifestEntriesCacheKey::ForExplicit("table", "path#main",
0);
ASSERT_EQ(CacheKind::SNAPSHOT_LIVE_MANIFEST, main_key->GetKind());
ASSERT_TRUE(CacheKeyEqual()(main_key, same_key));
diff --git a/src/paimon/core/bucket/bucket_select_converter.cpp
b/src/paimon/core/bucket/bucket_select_converter.cpp
index 3e8aa17e..5ae2b243 100644
--- a/src/paimon/core/bucket/bucket_select_converter.cpp
+++ b/src/paimon/core/bucket/bucket_select_converter.cpp
@@ -19,6 +19,7 @@
#include "paimon/core/bucket/bucket_select_converter.h"
#include <cassert>
+#include <cmath>
#include <set>
#include <string>
#include <utility>
@@ -34,6 +35,7 @@
#include "paimon/core/bucket/default_bucket_function.h"
#include "paimon/core/bucket/hive_bucket_function.h"
#include "paimon/core/bucket/mod_bucket_function.h"
+#include "paimon/core/casting/decimal_to_decimal_cast_executor.h"
#include "paimon/data/timestamp.h"
#include "paimon/memory/memory_pool.h"
#include "paimon/predicate/leaf_predicate.h"
@@ -46,9 +48,25 @@ Result<std::optional<int32_t>>
BucketSelectConverter::Convert(
const std::shared_ptr<Predicate>& predicate, const
std::vector<std::string>& bucket_key_names,
const std::vector<std::shared_ptr<arrow::DataType>>&
bucket_key_arrow_types,
BucketFunctionType bucket_function_type, int32_t num_buckets, MemoryPool*
pool) {
+ if (num_buckets <= 0) {
+ return std::optional<int32_t>();
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<BucketSelector> selector,
+ ConvertToSelector(predicate, bucket_key_names,
bucket_key_arrow_types,
+ bucket_function_type, pool));
+ if (!selector) {
+ return std::optional<int32_t>();
+ }
+ return std::optional<int32_t>(selector->Bucket(num_buckets));
+}
+
+Result<std::unique_ptr<BucketSelector>>
BucketSelectConverter::ConvertToSelector(
+ const std::shared_ptr<Predicate>& predicate, const
std::vector<std::string>& bucket_key_names,
+ const std::vector<std::shared_ptr<arrow::DataType>>&
bucket_key_arrow_types,
+ BucketFunctionType bucket_function_type, MemoryPool* pool) {
assert(pool);
- if (!predicate || bucket_key_names.empty() || num_buckets <= 0) {
- return std::optional<int32_t>(std::nullopt);
+ if (!predicate || bucket_key_names.empty()) {
+ return std::unique_ptr<BucketSelector>();
}
if (bucket_key_names.size() != bucket_key_arrow_types.size()) {
@@ -66,7 +84,7 @@ Result<std::optional<int32_t>> BucketSelectConverter::Convert(
auto literals_opt = ExtractEqualLiterals(predicate, bucket_key_names);
if (!literals_opt.has_value()) {
- return std::optional<int32_t>(std::nullopt);
+ return std::unique_ptr<BucketSelector>();
}
const auto& literals_map = literals_opt.value();
@@ -79,18 +97,31 @@ Result<std::optional<int32_t>>
BucketSelectConverter::Convert(
for (int32_t i = 0; i < num_fields; i++) {
const auto& field_name = bucket_key_names[i];
- const auto& literal = literals_map.at(field_name);
+ Literal literal = literals_map.at(field_name);
+ // Equal NaNs can have different stored bit patterns and therefore
different buckets.
+ if ((bucket_key_types[i] == FieldType::FLOAT &&
std::isnan(literal.GetValue<float>())) ||
+ (bucket_key_types[i] == FieldType::DOUBLE &&
std::isnan(literal.GetValue<double>()))) {
+ return std::unique_ptr<BucketSelector>();
+ }
+ if (bucket_key_types[i] == FieldType::DECIMAL) {
+ PAIMON_ASSIGN_OR_RAISE(Literal scaled,
DecimalToDecimalCastExecutor().Cast(
+ literal,
bucket_key_arrow_types[i]));
+ // Hash the field's representation only when rescaling preserves
the exact value.
+ if (scaled.IsNull() || !(scaled.GetValue<Decimal>() ==
literal.GetValue<Decimal>())) {
+ return std::unique_ptr<BucketSelector>();
+ }
+ literal = std::move(scaled);
+ }
PAIMON_RETURN_NOT_OK(
WriteLiteralToRow(i, literal, bucket_key_types[i],
bucket_key_arrow_types[i], &writer));
}
writer.Complete();
- // Create the bucket function and compute the bucket
+ // Retain the key and function; the bucket count belongs to each manifest
entry.
PAIMON_ASSIGN_OR_RAISE(
std::unique_ptr<BucketFunction> bucket_function,
CreateBucketFunction(bucket_function_type, bucket_key_types,
bucket_key_arrow_types));
- int32_t bucket = bucket_function->Bucket(row, num_buckets);
- return std::optional<int32_t>(bucket);
+ return std::make_unique<BucketSelector>(std::move(row),
std::move(bucket_function));
}
std::optional<std::map<std::string, Literal>>
BucketSelectConverter::ExtractEqualLiterals(
diff --git a/src/paimon/core/bucket/bucket_select_converter.h
b/src/paimon/core/bucket/bucket_select_converter.h
index 25eb31aa..e6921c39 100644
--- a/src/paimon/core/bucket/bucket_select_converter.h
+++ b/src/paimon/core/bucket/bucket_select_converter.h
@@ -23,10 +23,13 @@
#include <memory>
#include <optional>
#include <string>
+#include <utility>
#include <vector>
#include "arrow/type_fwd.h"
#include "paimon/bucket/bucket_function_type.h"
+#include "paimon/common/data/binary_row.h"
+#include "paimon/core/bucket/bucket_function.h"
#include "paimon/defs.h"
#include "paimon/predicate/literal.h"
#include "paimon/result.h"
@@ -38,6 +41,22 @@ class BucketFunction;
class MemoryPool;
class Predicate;
+/// Selects a bucket using the bucket count recorded in each manifest entry.
+class BucketSelector {
+ public:
+ BucketSelector(BinaryRow key, std::unique_ptr<BucketFunction> function)
+ : key_(std::move(key)), function_(std::move(function)) {}
+
+ /// Compute the bucket for a positive total bucket count.
+ int32_t Bucket(int32_t num_buckets) const {
+ return function_->Bucket(key_, num_buckets);
+ }
+
+ private:
+ BinaryRow key_;
+ std::unique_ptr<BucketFunction> function_;
+};
+
/// Converts predicates on bucket key fields to a target bucket ID.
/// When all bucket key fields have EQUAL predicates, the converter computes
/// which bucket the data must reside in, enabling bucket pruning during scan.
@@ -62,6 +81,14 @@ class BucketSelectConverter {
const std::vector<std::shared_ptr<arrow::DataType>>&
bucket_key_arrow_types,
BucketFunctionType bucket_function_type, int32_t num_buckets,
MemoryPool* pool);
+ /// Build a selector once, then evaluate it with each file's bucket count.
+ /// Returns nullptr when the predicate cannot safely constrain all bucket
keys.
+ static Result<std::unique_ptr<BucketSelector>> ConvertToSelector(
+ const std::shared_ptr<Predicate>& predicate,
+ const std::vector<std::string>& bucket_key_names,
+ const std::vector<std::shared_ptr<arrow::DataType>>&
bucket_key_arrow_types,
+ BucketFunctionType bucket_function_type, MemoryPool* pool);
+
private:
/// Extract single literal per bucket key field from EQUAL predicates.
/// Splits the predicate by AND and looks for EQUAL leaf predicates on
bucket key fields.
diff --git a/src/paimon/core/bucket/bucket_select_converter_test.cpp
b/src/paimon/core/bucket/bucket_select_converter_test.cpp
index 2707eccb..85a3f3e2 100644
--- a/src/paimon/core/bucket/bucket_select_converter_test.cpp
+++ b/src/paimon/core/bucket/bucket_select_converter_test.cpp
@@ -18,6 +18,7 @@
#include "paimon/core/bucket/bucket_select_converter.h"
+#include <limits>
#include <optional>
#include <string>
#include <vector>
@@ -60,6 +61,30 @@ class BucketSelectConverterTest : public ::testing::Test {
std::shared_ptr<MemoryPool> pool_ = GetDefaultPool();
};
+TEST_F(BucketSelectConverterTest, SelectorUsesEachBucketCount) {
+ auto pool = GetDefaultPool();
+ auto predicate =
+ PredicateBuilder::Equal(0, "key", FieldType::INT,
Literal(static_cast<int32_t>(-23)));
+ BinaryRow row =
BinaryRowGenerator::GenerateRow({static_cast<int32_t>(-23)}, pool.get());
+ ASSERT_OK_AND_ASSIGN(auto mod_function,
ModBucketFunction::Create(FieldType::INT));
+ ASSERT_OK_AND_ASSIGN(auto hive_function,
+
HiveBucketFunction::Create({HiveFieldInfo(FieldType::INT)}));
+ DefaultBucketFunction default_function;
+ const std::map<BucketFunctionType, const BucketFunction*> functions = {
+ {BucketFunctionType::DEFAULT, &default_function},
+ {BucketFunctionType::MOD, mod_function.get()},
+ {BucketFunctionType::HIVE, hive_function.get()}};
+ for (const auto& [type, function] : functions) {
+ ASSERT_OK_AND_ASSIGN(auto selector,
+ BucketSelectConverter::ConvertToSelector(
+ predicate, {"key"}, {arrow::int32()}, type,
pool.get()));
+ ASSERT_TRUE(selector);
+ for (int32_t total_buckets : {2, 4, 8, 17, 2}) {
+ ASSERT_EQ(selector->Bucket(total_buckets), function->Bucket(row,
total_buckets));
+ }
+ }
+}
+
TEST_F(BucketSelectConverterTest, SingleStringEqualDefault) {
std::string value = "hello_world";
AssertDefaultBucket(FieldType::STRING, Literal(FieldType::STRING,
value.c_str(), value.size()),
@@ -246,6 +271,55 @@ TEST_F(BucketSelectConverterTest,
HiveBucketFunctionWithDecimal) {
ASSERT_EQ(function->Bucket(row, num_buckets), selected_bucket.value());
}
+TEST_F(BucketSelectConverterTest, RescalesDecimalLiteralsExactly) {
+ for (int32_t precision : {10, 20}) {
+ for (BucketFunctionType function_type :
+ {BucketFunctionType::DEFAULT, BucketFunctionType::HIVE}) {
+ Decimal stored = Decimal::FromUnscaledLong(120, precision, 2);
+ auto stored_predicate =
+ PredicateBuilder::Equal(0, "amount", FieldType::DECIMAL,
Literal(stored));
+ ASSERT_OK_AND_ASSIGN(auto expected,
+
BucketSelectConverter::Convert(stored_predicate, {"amount"},
+
{arrow::decimal128(precision, 2)},
+ function_type,
17, pool_.get()));
+ ASSERT_TRUE(expected.has_value());
+ for (const auto& query : {Decimal::FromUnscaledLong(12, precision,
1),
+ Decimal::FromUnscaledLong(1200,
precision, 3)}) {
+ auto predicate =
+ PredicateBuilder::Equal(0, "amount", FieldType::DECIMAL,
Literal(query));
+ ASSERT_OK_AND_ASSIGN(
+ auto result, BucketSelectConverter::Convert(predicate,
{"amount"},
+
{arrow::decimal128(precision, 2)},
+ function_type,
17, pool_.get()));
+ ASSERT_EQ(result, expected);
+ }
+ }
+ }
+}
+
+TEST_F(BucketSelectConverterTest, InexactDecimalConversionReturnsNullopt) {
+ for (const auto& value :
+ {Decimal::FromUnscaledLong(123, 10, 3),
Decimal::FromUnscaledLong(9999999999LL, 10, 0)}) {
+ auto predicate = PredicateBuilder::Equal(0, "amount",
FieldType::DECIMAL, Literal(value));
+ ASSERT_OK_AND_ASSIGN(auto result, BucketSelectConverter::Convert(
+ predicate, {"amount"},
{arrow::decimal128(10, 2)},
+ BucketFunctionType::DEFAULT, 17,
pool_.get()));
+ ASSERT_FALSE(result.has_value());
+ }
+}
+
+TEST_F(BucketSelectConverterTest, NaNReturnsNullopt) {
+ for (const auto& value : {Literal(std::numeric_limits<float>::quiet_NaN()),
+
Literal(std::numeric_limits<double>::quiet_NaN())}) {
+ auto type = value.GetType() == FieldType::FLOAT ? arrow::float32() :
arrow::float64();
+ auto predicate = PredicateBuilder::Equal(0, "key", value.GetType(),
value);
+ ASSERT_OK_AND_ASSIGN(auto result, BucketSelectConverter::Convert(
+ predicate, {"key"}, {type},
+ BucketFunctionType::DEFAULT, 17,
pool_.get()));
+ ASSERT_FALSE(result.has_value());
+ }
+}
+
TEST_F(BucketSelectConverterTest, UnsupportedFieldTypeReturnsError) {
auto predicate =
PredicateBuilder::Equal(0, "items", FieldType::ARRAY,
Literal(static_cast<int32_t>(42)));
diff --git a/src/paimon/core/operation/append_only_file_store_scan.cpp
b/src/paimon/core/operation/append_only_file_store_scan.cpp
index 953bcadb..435ece57 100644
--- a/src/paimon/core/operation/append_only_file_store_scan.cpp
+++ b/src/paimon/core/operation/append_only_file_store_scan.cpp
@@ -26,6 +26,7 @@
#include <set>
#include <string>
#include <utility>
+#include <vector>
#include "arrow/type.h"
#include "fmt/format.h"
@@ -42,6 +43,7 @@
#include "paimon/core/utils/field_mapping.h"
#include "paimon/file_index/file_index_result.h"
#include "paimon/predicate/predicate_utils.h"
+#include "paimon/scan_context.h"
#include "paimon/status.h"
namespace paimon {
diff --git a/src/paimon/core/operation/append_only_file_store_scan.h
b/src/paimon/core/operation/append_only_file_store_scan.h
index 65ed66da..ba4feb1f 100644
--- a/src/paimon/core/operation/append_only_file_store_scan.h
+++ b/src/paimon/core/operation/append_only_file_store_scan.h
@@ -18,6 +18,7 @@
#pragma once
+#include <cstdint>
#include <memory>
#include <vector>
diff --git a/src/paimon/core/operation/append_only_file_store_scan_test.cpp
b/src/paimon/core/operation/append_only_file_store_scan_test.cpp
index fd16d47f..b45a5684 100644
--- a/src/paimon/core/operation/append_only_file_store_scan_test.cpp
+++ b/src/paimon/core/operation/append_only_file_store_scan_test.cpp
@@ -20,15 +20,19 @@
#include <algorithm>
#include <cstdint>
+#include <map>
#include <optional>
#include <string>
#include <utility>
#include <vector>
+#include "arrow/type.h"
#include "gtest/gtest.h"
#include "paimon/common/data/binary_row.h"
#include "paimon/common/data/binary_row_writer.h"
#include "paimon/common/io/cache/lru_cache.h"
+#include "paimon/common/utils/math.h"
+#include "paimon/core/bucket/default_bucket_function.h"
#include "paimon/core/manifest/manifest_entry.h"
#include "paimon/core/manifest/partition_entry.h"
#include "paimon/core/schema/schema_manager.h"
@@ -37,6 +41,7 @@
#include "paimon/core/table/source/abstract_table_scan.h"
#include "paimon/core/table/source/snapshot/snapshot_reader.h"
#include "paimon/defs.h"
+#include "paimon/executor.h"
#include "paimon/fs/local/local_file_system.h"
#include "paimon/memory/memory_pool.h"
#include "paimon/metrics.h"
@@ -46,10 +51,168 @@
#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,
+ CreateDefaultExecutor(), 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();
+ auto query_value =
FloatingPointFromBits<double>(uint64_t{0x7ff8000000000000ULL});
+ auto stored_value =
FloatingPointFromBits<double>(uint64_t{0x7ff8000000000001ULL});
+ CheckMatchingValue(FieldType::DOUBLE, query_value, stored_value);
+}
+
TEST(AppendOnlyFileStoreScanTest, TestReconstructPredicateWithNonCastedFields)
{
std::string table_root =
paimon::test::GetDataDir() +
diff --git a/src/paimon/core/operation/file_store_scan.cpp
b/src/paimon/core/operation/file_store_scan.cpp
index befe7eb9..2b182107 100644
--- a/src/paimon/core/operation/file_store_scan.cpp
+++ b/src/paimon/core/operation/file_store_scan.cpp
@@ -33,6 +33,7 @@
#include "paimon/common/data/binary_array.h"
#include "paimon/common/data/blob_utils.h"
#include "paimon/common/executor/future.h"
+#include "paimon/common/io/cache/cache_key.h"
#include "paimon/common/predicate/literal_converter.h"
#include "paimon/common/types/data_field.h"
#include "paimon/common/utils/field_type_utils.h"
@@ -146,16 +147,20 @@ Result<std::shared_ptr<FileStoreScan::RawPlan>>
FileStoreScan::CreatePlan() cons
ReadManifests(&snapshot, &all_manifest_file_metas,
&filtered_manifest_file_metas));
std::vector<ManifestEntry> manifest_entries;
+ std::optional<int32_t> cache_bucket = bucket_filter_;
+ if (!cache_bucket && bucket_selector_) {
+ cache_bucket = bucket_selector_->Bucket(core_options_.GetBucket());
+ }
const bool use_snapshot_live_manifest_cache =
snapshot.has_value() && scan_mode_ == ScanMode::ALL &&
core_options_.GetScanManifestEntryCacheMaxSnapshots() > 0 &&
core_options_.GetCache() != nullptr && !table_path_.empty() &&
- !row_range_index_.has_value() && bucket_filter_.has_value();
+ !row_range_index_.has_value() && cache_bucket.has_value();
uint64_t lazy_decode_scanned_rows = 0;
bool snapshot_cache_hit = false;
if (use_snapshot_live_manifest_cache) {
PAIMON_RETURN_NOT_OK(ReadManifestEntriesWithCache(snapshot.value(),
all_manifest_file_metas,
-
bucket_filter_.value(), &manifest_entries,
+
cache_bucket.value(), &manifest_entries,
&snapshot_cache_hit));
lazy_decode_scanned_rows = manifest_entries.size();
std::vector<ManifestEntry> filtered_entries;
@@ -325,7 +330,8 @@ Status FileStoreScan::ReadManifestEntries(const
std::vector<ManifestFileMeta>& m
}
// Cache merged live manifest entries for one bucket before applying scan
filters. Each cache value
-// keeps a bounded number of snapshot results for the same
table/branch/bucket. Exact snapshot hits
+// keeps bounded snapshot results for a table/branch/bucket. Inferred entries
also include other
+// bucket counts and schema IDs, with the current count and schema in the
cache key. Exact hits
// can be returned directly; cache misses rebuild the target snapshot bucket
from the target
// snapshot's data manifests.
Status FileStoreScan::ReadManifestEntriesWithCache(
@@ -351,7 +357,7 @@ Status FileStoreScan::ReadManifestEntriesWithCache(
// cache.
std::vector<ManifestFileMeta> bucket_manifest_metas;
for (const auto& meta : all_manifest_metas) {
- if (MayContainBucket(meta, bucket)) {
+ if ((!bucket_filter_ && bucket_selector_) || MayContainBucket(meta,
bucket)) {
bucket_manifest_metas.push_back(meta);
}
}
@@ -368,8 +374,14 @@ Status FileStoreScan::ReadManifestEntriesWithCache(
return Status::OK();
}
-std::shared_ptr<CacheKey>
FileStoreScan::SnapshotLiveManifestEntriesCacheKey(int32_t bucket) const {
- return CacheKey::ForSnapshotLiveManifestEntries(
+std::shared_ptr<CacheKey>
FileStoreScan::CreateSnapshotLiveManifestEntriesCacheKey(
+ int32_t bucket) const {
+ if (!bucket_filter_ && bucket_selector_) {
+ return SnapshotLiveManifestEntriesCacheKey::ForInferred(
+ table_path_,
BranchManager::NormalizeBranch(core_options_.GetBranch()), bucket,
+ core_options_.GetBucket(), table_schema_->Id());
+ }
+ return SnapshotLiveManifestEntriesCacheKey::ForExplicit(
table_path_,
BranchManager::NormalizeBranch(core_options_.GetBranch()), bucket);
}
@@ -378,7 +390,7 @@ Result<SnapshotLiveManifestEntries>
FileStoreScan::LoadSnapshotLiveManifestEntri
auto supplier = [](const std::shared_ptr<CacheKey>&) ->
Result<std::shared_ptr<CacheValue>> {
return std::shared_ptr<CacheValue>();
};
- std::shared_ptr<CacheKey> cache_key =
SnapshotLiveManifestEntriesCacheKey(bucket);
+ std::shared_ptr<CacheKey> cache_key =
CreateSnapshotLiveManifestEntriesCacheKey(bucket);
const auto max_snapshots =
core_options_.GetScanManifestEntryCacheMaxSnapshots();
Result<std::shared_ptr<CacheValue>> cache_result =
core_options_.GetCache()->Get(cache_key, supplier);
@@ -401,15 +413,17 @@ Status FileStoreScan::StoreSnapshotLiveManifestEntries(
}
auto cache_value =
std::make_shared<CacheValue>(MemorySegment::Wrap(bytes_result.value()),
CacheCallback());
- Status status =
-
core_options_.GetCache()->Put(SnapshotLiveManifestEntriesCacheKey(bucket),
cache_value);
+ Status status =
core_options_.GetCache()->Put(CreateSnapshotLiveManifestEntriesCacheKey(bucket),
+ cache_value);
return status.ok() ? status : Status::OK();
}
Status FileStoreScan::ReadAndMergeBucketFileEntries(
const std::vector<ManifestFileMeta>& manifest_metas, int32_t bucket,
std::vector<ManifestEntry>* merged_entries) const {
- if (core_options_.ScanManifestEntryLazyDecodeEnabled()) {
+ const bool inferred_bucket = !bucket_filter_ && bucket_selector_ !=
nullptr;
+ // Explicit-bucket lazy decoding cannot retain entries with a different
layout.
+ if (!inferred_bucket &&
core_options_.ScanManifestEntryLazyDecodeEnabled()) {
std::vector<std::future<Result<std::vector<ManifestEntry>>>> futures;
futures.reserve(manifest_metas.size());
for (const auto& meta : manifest_metas) {
@@ -441,7 +455,9 @@ Status FileStoreScan::ReadAndMergeBucketFileEntries(
PAIMON_RETURN_NOT_OK(ReadFileEntries(manifest_metas, &entries,
/*apply_scan_filter=*/false));
unmerged_entries.reserve(entries.size());
for (auto& entry : entries) {
- if (entry.Bucket() == bucket) {
+ if (entry.Bucket() == bucket ||
+ (inferred_bucket && (entry.TotalBuckets() !=
core_options_.GetBucket() ||
+ entry.File()->schema_id !=
table_schema_->Id()))) {
unmerged_entries.emplace_back(std::move(entry));
}
}
@@ -557,12 +573,43 @@ Result<bool> FileStoreScan::FilterManifestEntry(const
ManifestEntry& entry) cons
if (bucket_filter_ != std::nullopt && entry.Bucket() !=
bucket_filter_.value()) {
return false;
}
+ if (bucket_selector_ && entry.TotalBuckets() > 0) {
+ PAIMON_ASSIGN_OR_RAISE(bool compatible,
HasCompatibleBucketKeys(entry.File()->schema_id));
+ if (compatible && entry.Bucket() !=
bucket_selector_->Bucket(entry.TotalBuckets())) {
+ return false;
+ }
+ }
if (level_filter_ != nullptr && !level_filter_(entry.Level())) {
return false;
}
return FilterByStats(entry);
}
+Result<bool> FileStoreScan::HasCompatibleBucketKeys(int64_t data_schema_id)
const {
+ if (data_schema_id == table_schema_->Id()) {
+ return true;
+ }
+ auto cached = bucket_schema_compatibility_.Find(data_schema_id);
+ if (cached) {
+ return cached.value();
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<TableSchema> data_schema,
+ schema_manager_->ReadSchema(data_schema_id));
+ const auto& current_keys = table_schema_->BucketKeys();
+ const auto& data_keys = data_schema->BucketKeys();
+ PAIMON_ASSIGN_OR_RAISE(CoreOptions data_options,
CoreOptions::FromMap(data_schema->Options()));
+ bool compatible = current_keys.size() == data_keys.size() &&
+ core_options_.GetBucketFunctionType() ==
data_options.GetBucketFunctionType();
+ for (size_t i = 0; compatible && i < current_keys.size(); ++i) {
+ PAIMON_ASSIGN_OR_RAISE(DataField current_field,
table_schema_->GetField(current_keys[i]));
+ PAIMON_ASSIGN_OR_RAISE(DataField data_field,
data_schema->GetField(data_keys[i]));
+ compatible = current_field.Id() == data_field.Id() &&
+ current_field.Type()->Equals(data_field.Type());
+ }
+ bucket_schema_compatibility_.Insert(data_schema_id, compatible);
+ return compatible;
+}
+
Status FileStoreScan::SplitAndSetFilter(const std::vector<std::string>&
partition_keys,
const std::shared_ptr<arrow::Schema>&
arrow_schema,
const std::shared_ptr<ScanFilter>&
scan_filters) {
@@ -602,6 +649,19 @@ Status FileStoreScan::SplitAndSetFilter(const
std::vector<std::string>& partitio
}
}
bucket_filter_ = scan_filters->GetBucketFilter();
+ const auto& bucket_keys = table_schema_->BucketKeys();
+ if (predicates_ && !bucket_filter_ && core_options_.GetBucket() > 0 &&
!bucket_keys.empty()) {
+ std::vector<std::shared_ptr<arrow::DataType>> bucket_key_types;
+ bucket_key_types.reserve(bucket_keys.size());
+ for (const auto& key : bucket_keys) {
+ PAIMON_ASSIGN_OR_RAISE(DataField field,
table_schema_->GetField(key));
+ bucket_key_types.push_back(field.Type());
+ }
+ PAIMON_ASSIGN_OR_RAISE(bucket_selector_,
+ BucketSelectConverter::ConvertToSelector(
+ predicates_, bucket_keys, bucket_key_types,
+ core_options_.GetBucketFunctionType(),
pool_.get()));
+ }
if (!scan_filters->GetPartitionFilters().empty()) {
PAIMON_ASSIGN_OR_RAISE(
partition_filter_,
diff --git a/src/paimon/core/operation/file_store_scan.h
b/src/paimon/core/operation/file_store_scan.h
index ea963a71..83c18d49 100644
--- a/src/paimon/core/operation/file_store_scan.h
+++ b/src/paimon/core/operation/file_store_scan.h
@@ -34,8 +34,10 @@
#include "paimon/common/predicate/leaf_predicate_impl.h"
#include "paimon/common/predicate/literal_converter.h"
#include "paimon/common/predicate/predicate_filter.h"
+#include "paimon/common/utils/concurrent_hash_map.h"
#include "paimon/common/utils/field_type_utils.h"
#include "paimon/common/utils/linked_hash_map.h"
+#include "paimon/core/bucket/bucket_select_converter.h"
#include "paimon/core/core_options.h"
#include "paimon/core/manifest/manifest_entry.h"
#include "paimon/core/manifest/manifest_file.h"
@@ -241,14 +243,6 @@ class FileStoreScan {
const std::shared_ptr<arrow::Schema>&
arrow_schema,
const std::shared_ptr<ScanFilter>& scan_filters);
- /// Set the bucket filter derived from predicate analysis (e.g.,
BucketSelectConverter).
- /// Only sets the filter if no explicit bucket filter was already set.
- void SetBucketFilterIfAbsent(int32_t bucket) {
- if (!bucket_filter_.has_value()) {
- bucket_filter_ = bucket;
- }
- }
-
// When schema evolves, predicates might contain fields requiring casting.
To avoid false
// negatives when filtering by stats, we exclude those fields from
predicate.
static Result<std::shared_ptr<Predicate>>
ReconstructPredicateWithNonCastedFields(
@@ -271,7 +265,7 @@ class FileStoreScan {
int32_t bucket,
std::vector<ManifestEntry>*
manifest_entries,
bool* cache_hit) const;
- std::shared_ptr<CacheKey> SnapshotLiveManifestEntriesCacheKey(int32_t
bucket) const;
+ std::shared_ptr<CacheKey>
CreateSnapshotLiveManifestEntriesCacheKey(int32_t bucket) const;
Result<SnapshotLiveManifestEntries>
LoadSnapshotLiveManifestEntries(int32_t bucket) const;
Status StoreSnapshotLiveManifestEntries(int32_t bucket,
const SnapshotLiveManifestEntries&
entries) const;
@@ -300,6 +294,7 @@ class FileStoreScan {
std::vector<ManifestEntry>* entries) const;
Result<bool> FilterManifestEntry(const ManifestEntry& entry) const;
+ Result<bool> HasCompatibleBucketKeys(int64_t data_schema_id) const;
protected:
std::shared_ptr<MemoryPool> pool_;
@@ -321,6 +316,8 @@ class FileStoreScan {
std::shared_ptr<PredicateFilter> partition_filter_;
std::shared_ptr<Executor> executor_;
std::optional<int32_t> bucket_filter_;
+ std::unique_ptr<BucketSelector> bucket_selector_;
+ mutable ConcurrentHashMap<int64_t, bool> bucket_schema_compatibility_;
std::function<bool(int32_t)> level_filter_;
std::optional<Snapshot> specified_snapshot_;
std::shared_ptr<Metrics> metrics_;
diff --git a/src/paimon/core/operation/key_value_file_store_scan.cpp
b/src/paimon/core/operation/key_value_file_store_scan.cpp
index 54af98a0..c8b8bacc 100644
--- a/src/paimon/core/operation/key_value_file_store_scan.cpp
+++ b/src/paimon/core/operation/key_value_file_store_scan.cpp
@@ -31,7 +31,6 @@
#include "paimon/common/predicate/predicate_filter.h"
#include "paimon/common/types/data_field.h"
#include "paimon/common/utils/object_utils.h"
-#include "paimon/core/bucket/bucket_select_converter.h"
#include "paimon/core/core_options.h"
#include "paimon/core/io/data_file_meta.h"
#include "paimon/core/options/merge_engine.h"
@@ -121,26 +120,6 @@ Status KeyValueFileStoreScan::SplitAndSetKeyValueFilter(
return Status::Invalid("invalid key predicate, cannot cast to
PredicateFilter");
}
WithKeyFilter(key_filter);
-
- // Bucket select conversion: derive target bucket from EQUAL
predicates on bucket keys
- const auto& bucket_keys = table_schema_->BucketKeys();
- int32_t num_buckets = core_options_.GetBucket();
- if (num_buckets > 0 && !bucket_keys.empty()) {
- std::vector<std::shared_ptr<arrow::DataType>>
bucket_key_arrow_types;
- bucket_key_arrow_types.reserve(bucket_keys.size());
- for (const auto& key : bucket_keys) {
- PAIMON_ASSIGN_OR_RAISE(DataField field,
table_schema_->GetField(key));
- bucket_key_arrow_types.push_back(field.Type());
- }
- PAIMON_ASSIGN_OR_RAISE(
- std::optional<int32_t> selected_bucket,
- BucketSelectConverter::Convert(key_predicate, bucket_keys,
bucket_key_arrow_types,
-
core_options_.GetBucketFunctionType(), num_buckets,
- pool_.get()));
- if (selected_bucket.has_value()) {
- SetBucketFilterIfAbsent(selected_bucket.value());
- }
- }
}
// Only set value filtering when there are predicates on non-primary-key
fields.
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 f6c13cb3..ae53d8aa 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
@@ -27,6 +27,7 @@
#include "paimon/common/data/binary_row.h"
#include "paimon/common/predicate/predicate_filter.h"
#include "paimon/common/types/data_field.h"
+#include "paimon/core/bucket/default_bucket_function.h"
#include "paimon/core/core_options.h"
#include "paimon/core/io/data_file_meta.h"
#include "paimon/core/manifest/file_kind.h"
@@ -44,6 +45,7 @@
#include "paimon/defs.h"
#include "paimon/executor.h"
#include "paimon/format/file_format.h"
+#include "paimon/fs/local/local_file_system.h"
#include "paimon/memory/memory_pool.h"
#include "paimon/metrics.h"
#include "paimon/predicate/literal.h"
@@ -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,
CreateDefaultExecutor(), pool));
+ SimpleStats stats =
BinaryRowGenerator::GenerateStats({std::string("a")},
+
{std::string("z")}, {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));
+ file->key_stats = stats;
+ BinaryRow key = BinaryRowGenerator::GenerateRow({std::string("key")},
pool.get());
+ for (int32_t total_buckets : {2, 4, 8, 17}) {
+ int32_t expected = DefaultBucketFunction().Bucket(key,
total_buckets);
+ for (int32_t bucket = 0; bucket < total_buckets; ++bucket) {
+ ManifestEntry entry(FileKind::Add(), BinaryRow::EmptyRow(),
bucket, total_buckets,
+ file);
+ ASSERT_OK_AND_ASSIGN(bool stats_match,
scan->FilterByStats(entry));
+ ASSERT_TRUE(stats_match);
+ ASSERT_OK_AND_ASSIGN(bool keep,
scan->FilterManifestEntry(entry));
+ ASSERT_EQ(keep, bucket == expected)
+ << "bucket=" << bucket << " total=" << total_buckets;
+ }
+ }
+ }
+}
+
class KeyValueFileStoreScanTest : public testing::Test {
public:
void SetUp() override {
diff --git a/src/paimon/core/schema/schema_manager.cpp
b/src/paimon/core/schema/schema_manager.cpp
index ede262f2..4ec1f8ec 100644
--- a/src/paimon/core/schema/schema_manager.cpp
+++ b/src/paimon/core/schema/schema_manager.cpp
@@ -71,16 +71,16 @@ Result<std::optional<std::shared_ptr<TableSchema>>>
SchemaManager::Latest() cons
}
Result<std::shared_ptr<TableSchema>> SchemaManager::ReadSchema(int64_t
schema_id) const {
- auto iter = schema_cache_.find(schema_id);
- if (iter != schema_cache_.end()) {
- return iter->second;
+ auto cached = schema_cache_.Find(schema_id);
+ if (cached) {
+ return cached.value();
}
auto path = ToSchemaPath(schema_id);
std::string content;
PAIMON_RETURN_NOT_OK(file_system_->ReadFile(path, &content));
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<TableSchema> schema,
TableSchema::CreateFromJson(content));
- schema_cache_[schema_id] = schema;
+ schema_cache_.Insert(schema_id, schema);
return schema;
}
diff --git a/src/paimon/core/schema/schema_manager.h
b/src/paimon/core/schema/schema_manager.h
index 382fa303..7480ec5d 100644
--- a/src/paimon/core/schema/schema_manager.h
+++ b/src/paimon/core/schema/schema_manager.h
@@ -26,6 +26,7 @@
#include <string>
#include <vector>
+#include "paimon/common/utils/concurrent_hash_map.h"
#include "paimon/core/schema/table_schema.h"
#include "paimon/fs/file_system.h"
#include "paimon/result.h"
@@ -69,7 +70,7 @@ class SchemaManager {
std::shared_ptr<FileSystem> file_system_;
std::string table_root_;
const std::string branch_;
- mutable std::map<int64_t, std::shared_ptr<TableSchema>> schema_cache_;
+ mutable ConcurrentHashMap<int64_t, std::shared_ptr<TableSchema>>
schema_cache_;
};
} // namespace paimon
diff --git a/src/paimon/core/schema/schema_manager_test.cpp
b/src/paimon/core/schema/schema_manager_test.cpp
index 83f995a9..0b078462 100644
--- a/src/paimon/core/schema/schema_manager_test.cpp
+++ b/src/paimon/core/schema/schema_manager_test.cpp
@@ -19,6 +19,7 @@
#include "paimon/core/schema/schema_manager.h"
+#include <future>
#include <set>
#include <utility>
@@ -30,6 +31,22 @@
namespace paimon::test {
+TEST(SchemaManagerTest, ConcurrentHistoricalSchemaReads) {
+ SchemaManager manager(
+ std::make_shared<LocalFileSystem>(),
+ GetDataDir() +
"/orc/pk_table_with_alter_table.db/pk_table_with_alter_table/");
+ std::vector<std::future<Result<std::shared_ptr<TableSchema>>>> reads;
+ for (int32_t i = 0; i < 16; ++i) {
+ reads.push_back(
+ std::async(std::launch::async, [&manager, i]() { return
manager.ReadSchema(i % 2); }));
+ }
+ for (int32_t i = 0; i < 16; ++i) {
+ ASSERT_OK_AND_ASSIGN(auto schema, reads[i].get());
+ ASSERT_EQ(schema->Id(), i % 2);
+ }
+ ASSERT_EQ(manager.schema_cache_.Size(), 2);
+}
+
TEST(SchemaManagerTest, TestSimple) {
auto fs = std::make_shared<LocalFileSystem>();
std::string table_root =
@@ -91,7 +108,7 @@ TEST(SchemaManagerTest, TestSimple) {
})";
ASSERT_OK_AND_ASSIGN(auto expected_schema,
TableSchema::CreateFromJson(schema_json));
ASSERT_EQ(*ret, *expected_schema);
- ASSERT_FALSE(manager.schema_cache_.empty());
+ ASSERT_GT(manager.schema_cache_.Size(), 0);
ASSERT_EQ(*manager.ReadSchema(/*schema_id=*/1).value(), *expected_schema);
ASSERT_EQ(*(manager.Latest().value().value()), *expected_schema);
}
diff --git a/test/inte/scan_and_read_inte_test.cpp
b/test/inte/scan_and_read_inte_test.cpp
index c4c50cc4..ad134540 100644
--- a/test/inte/scan_and_read_inte_test.cpp
+++ b/test/inte/scan_and_read_inte_test.cpp
@@ -29,8 +29,12 @@
#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"
@@ -56,6 +60,7 @@
#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"
@@ -421,6 +426,136 @@ TEST_P(ScanAndReadInteTest, TestWithAppendSnapshot3) {
ASSERT_TRUE(expected->Equals(read_result)) << read_result->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";
@@ -3025,46 +3160,68 @@ TEST_P(ScanAndReadInteTest,
TestWithPKBucketSelectByPredicate) {
auto predicate = PredicateBuilder::Equal(/*field_index=*/2,
/*field_name=*/"f2", FieldType::INT,
Literal(static_cast<int32_t>(0)));
- ScanContextBuilder scan_context_builder(table_path);
- scan_context_builder.AddOption(Options::SCAN_SNAPSHOT_ID, "6");
- 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)));
+ // 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)));
+ 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());
- ASSERT_EQ(result_plan->SnapshotId().value(), 6);
+ 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);
+ }
- // 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_EQ(result_plan->SnapshotId().value(), 6);
- ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(splits));
- ASSERT_OK_AND_ASSIGN(auto read_result,
-
ReadResultCollector::CollectResult(std::move(batch_reader)));
+ // 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);
+ }
- // 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"([
+ 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();
+ .ValueOrDie());
+ ASSERT_TRUE(expected);
+ ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString();
+ }
}
TEST_P(ScanAndReadInteTest, TestReadNullableMapKey) {
