Skip to content

Commit db094f0

Browse files
lucasfanglucasfang
committed
fix
1 parent 48b2d2d commit db094f0

8 files changed

Lines changed: 137 additions & 19 deletions

File tree

‎src/paimon/CMakeLists.txt‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -329,6 +329,7 @@ set(PAIMON_CORE_SRCS
329329
core/stats/simple_stats.cpp
330330
core/stats/simple_stats_evolution.cpp
331331
core/table/table.cpp
332+
core/table/bucket_mode.cpp
332333
core/table/sink/commit_message.cpp
333334
core/table/sink/commit_message_impl.cpp
334335
core/table/sink/commit_message_serializer.cpp
@@ -750,6 +751,7 @@ if(PAIMON_BUILD_TESTS)
750751
core/stats/simple_stats_collector_test.cpp
751752
core/stats/simple_stats_test.cpp
752753
core/table/table_test.cpp
754+
core/table/bucket_mode_test.cpp
753755
core/table/sink/commit_message_test.cpp
754756
core/table/sink/commit_message_impl_test.cpp
755757
core/table/source/fallback_data_split_test.cpp

‎src/paimon/core/operation/commit/conflict_detection.cpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -88,7 +88,7 @@ Status ConflictDetection::CheckConflicts(
8888
row_id_column_conflict_checker,
8989
const Snapshot::CommitKind& commit_kind) const {
9090
if (options_.DeletionVectorsEnabled() &&
91-
options_.GetBucket() == BucketModeDefine::UNAWARE_BUCKET) {
91+
ResolveBucketMode(options_.GetBucket(), table_schema_) == BucketMode::BUCKET_UNAWARE) {
9292
return Status::NotImplemented(
9393
"check conflicts failed. not yet support dv with BUCKET_UNAWARE mode");
9494
}

‎src/paimon/core/operation/commit/conflict_detection_test.cpp‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -427,6 +427,39 @@ TEST_F(ConflictDetectionTest, TestDeletionVectorsNotSupportedWithBucketUnawareMo
427427
"not yet support dv with BUCKET_UNAWARE mode");
428428
}
429429

430+
TEST_F(ConflictDetectionTest,
431+
TestDeletionVectorsNotSupportedWithResolvedBucketUnawareModeFromMinusOne) {
432+
ASSERT_OK_AND_ASSIGN(
433+
std::shared_ptr<TableSchema> table_schema,
434+
TableSchema::Create(/*schema_id=*/0, arrow::schema(fields_), /*partition_keys=*/{"f1"},
435+
/*primary_keys=*/{}, /*options=*/{}));
436+
ASSERT_OK_AND_ASSIGN(CoreOptions core_options,
437+
CoreOptions::FromMap({{Options::BUCKET, "-1"},
438+
{Options::DELETION_VECTORS_ENABLED, "true"}}));
439+
ConflictDetection detection(table_schema, core_options, nullptr, nullptr, nullptr);
440+
441+
ASSERT_NOK_WITH_MSG(CheckConflicts(detection, /*base_entries=*/{}, /*delta_entries=*/{},
442+
Snapshot::CommitKind::Append()),
443+
"not yet support dv with BUCKET_UNAWARE mode");
444+
}
445+
446+
TEST_F(ConflictDetectionTest, TestDeletionVectorsAllowedWithResolvedDynamicBucketMode) {
447+
auto fields = {arrow::field("f0", arrow::int32(), /*nullable=*/false),
448+
arrow::field("f1", arrow::int32(), /*nullable=*/false),
449+
arrow::field("f2", arrow::int32())};
450+
ASSERT_OK_AND_ASSIGN(
451+
std::shared_ptr<TableSchema> table_schema,
452+
TableSchema::Create(/*schema_id=*/0, arrow::schema(fields), /*partition_keys=*/{"f1"},
453+
/*primary_keys=*/{"f1", "f0"}, {{Options::BUCKET, "-1"}}));
454+
ASSERT_OK_AND_ASSIGN(CoreOptions core_options,
455+
CoreOptions::FromMap({{Options::BUCKET, "-1"},
456+
{Options::DELETION_VECTORS_ENABLED, "true"}}));
457+
ConflictDetection detection(table_schema, core_options, nullptr, nullptr, nullptr);
458+
459+
ASSERT_OK(CheckConflicts(detection, /*base_entries=*/{}, /*delta_entries=*/{},
460+
Snapshot::CommitKind::Append()));
461+
}
462+
430463
TEST_F(ConflictDetectionTest, TestCheckLsmKeyRangeConflict) {
431464
auto fields = {arrow::field("f0", arrow::int32(), /*nullable=*/false),
432465
arrow::field("f1", arrow::int32(), /*nullable=*/false),

‎src/paimon/core/operation/file_store_commit_impl.cpp‎

Lines changed: 0 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -812,21 +812,6 @@ bool FileStoreCommitImpl::ShouldCheckSameBucket(const Snapshot::CommitKind& comm
812812
(IsUnorderedWriteOnlyAppend() || IsWriteOnlySnapshotSequenceAppend());
813813
}
814814

815-
BucketMode FileStoreCommitImpl::ResolveBucketMode(
816-
int32_t bucket, const std::shared_ptr<TableSchema>& table_schema) {
817-
if (bucket == BucketModeDefine::POSTPONE_BUCKET) {
818-
return BucketMode::POSTPONE_MODE;
819-
}
820-
if (bucket == -1) {
821-
return table_schema->PrimaryKeys().empty() ? BucketMode::BUCKET_UNAWARE
822-
: BucketMode::HASH_DYNAMIC;
823-
}
824-
if (bucket == BucketModeDefine::UNAWARE_BUCKET) {
825-
return BucketMode::BUCKET_UNAWARE;
826-
}
827-
return BucketMode::HASH_FIXED;
828-
}
829-
830815
bool FileStoreCommitImpl::IsUnorderedWriteOnlyAppend() const {
831816
return options_.WriteOnly() && !options_.BucketAppendOrdered();
832817
}

‎src/paimon/core/operation/file_store_commit_impl.h‎

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -224,9 +224,6 @@ class FileStoreCommitImpl : public FileStoreCommit {
224224

225225
static int64_t RowCounts(const std::vector<ManifestEntry>& files);
226226

227-
static BucketMode ResolveBucketMode(int32_t bucket,
228-
const std::shared_ptr<TableSchema>& table_schema);
229-
230227
private:
231228
std::shared_ptr<MemoryPool> memory_pool_;
232229
std::shared_ptr<Executor> executor_;

‎src/paimon/core/table/bucket_mode.cpp‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,37 @@
1+
/*
2+
* Copyright 2026-present Alibaba Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
#include "paimon/core/table/bucket_mode.h"
18+
19+
#include "paimon/core/schema/table_schema.h"
20+
21+
namespace paimon {
22+
23+
BucketMode ResolveBucketMode(int32_t bucket, const std::shared_ptr<TableSchema>& table_schema) {
24+
if (bucket == BucketModeDefine::POSTPONE_BUCKET) {
25+
return BucketMode::POSTPONE_MODE;
26+
}
27+
if (bucket == -1) {
28+
return table_schema->PrimaryKeys().empty() ? BucketMode::BUCKET_UNAWARE
29+
: BucketMode::HASH_DYNAMIC;
30+
}
31+
if (bucket == BucketModeDefine::UNAWARE_BUCKET) {
32+
return BucketMode::BUCKET_UNAWARE;
33+
}
34+
return BucketMode::HASH_FIXED;
35+
}
36+
37+
} // namespace paimon

‎src/paimon/core/table/bucket_mode.h‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,9 +17,12 @@
1717
#pragma once
1818

1919
#include <cstdint>
20+
#include <memory>
2021

2122
namespace paimon {
2223

24+
class TableSchema;
25+
2326
/// Bucket mode of the table, it affects the writing process and also affects the data skipping in
2427
/// reading.
2528
enum class BucketMode {
@@ -62,4 +65,6 @@ class BucketModeDefine {
6265
static constexpr int32_t POSTPONE_BUCKET = -2;
6366
};
6467

68+
BucketMode ResolveBucketMode(int32_t bucket, const std::shared_ptr<TableSchema>& table_schema);
69+
6570
} // namespace paimon

‎src/paimon/core/table/bucket_mode_test.cpp‎

Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,59 @@
1+
/*
2+
* Copyright 2026-present Alibaba Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
#include "paimon/core/table/bucket_mode.h"
18+
19+
#include <map>
20+
#include <memory>
21+
#include <string>
22+
#include <vector>
23+
24+
#include "arrow/api.h"
25+
#include "gtest/gtest.h"
26+
#include "paimon/core/schema/table_schema.h"
27+
#include "paimon/testing/utils/testharness.h"
28+
29+
namespace paimon::test {
30+
31+
namespace {
32+
33+
std::shared_ptr<TableSchema> CreateTableSchema(const std::vector<std::string>& primary_keys) {
34+
auto schema = arrow::schema({arrow::field("f0", arrow::int32(), /*nullable=*/false),
35+
arrow::field("f1", arrow::int32(), /*nullable=*/false)});
36+
std::vector<std::string> partition_keys = {"f1"};
37+
std::map<std::string, std::string> options;
38+
EXPECT_OK_AND_ASSIGN(
39+
std::shared_ptr<TableSchema> table_schema,
40+
TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
41+
return table_schema;
42+
}
43+
44+
} // namespace
45+
46+
TEST(BucketModeTest, TestResolveBucketMode) {
47+
std::shared_ptr<TableSchema> append_schema = CreateTableSchema(/*primary_keys=*/{});
48+
std::shared_ptr<TableSchema> pk_schema = CreateTableSchema(/*primary_keys=*/{"f0"});
49+
50+
EXPECT_EQ(BucketMode::POSTPONE_MODE,
51+
ResolveBucketMode(BucketModeDefine::POSTPONE_BUCKET, append_schema));
52+
EXPECT_EQ(BucketMode::BUCKET_UNAWARE, ResolveBucketMode(-1, append_schema));
53+
EXPECT_EQ(BucketMode::HASH_DYNAMIC, ResolveBucketMode(-1, pk_schema));
54+
EXPECT_EQ(BucketMode::BUCKET_UNAWARE,
55+
ResolveBucketMode(BucketModeDefine::UNAWARE_BUCKET, pk_schema));
56+
EXPECT_EQ(BucketMode::HASH_FIXED, ResolveBucketMode(4, append_schema));
57+
}
58+
59+
} // namespace paimon::test

0 commit comments

Comments
 (0)