Skip to content

Commit 328c789

Browse files
committed
fix
1 parent 6eb2720 commit 328c789

23 files changed

Lines changed: 708 additions & 117 deletions

include/paimon/defs.h

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -236,7 +236,7 @@ struct PAIMON_EXPORT Options {
236236
/// "commit.max-retry-wait" - Max retry wait time when commit failed. Default value is 10s.
237237
static const char COMMIT_MAX_RETRY_WAIT[];
238238

239-
/// "commit.discard-duplicate-files" - Whether to discard duplicate files on APPEND commit.
239+
/// "commit.discard-duplicate-files" - Whether to discard duplicate files in commit.
240240
/// Default value is "false".
241241
static const char COMMIT_DISCARD_DUPLICATE_FILES[];
242242

@@ -284,6 +284,10 @@ struct PAIMON_EXPORT Options {
284284
/// the table has partition keys. Default value is true.
285285
static const char DYNAMIC_PARTITION_OVERWRITE[];
286286

287+
/// "pk-clustering-override" - Whether to enable clustering by non-primary key fields.
288+
/// This option is currently not supported in paimon-cpp. Default value is false.
289+
static const char PK_CLUSTERING_OVERRIDE[];
290+
287291
/// "lookup-compact.max-interval" - The max interval for a gentle mode lookup compaction to be
288292
/// triggered. For every interval, a forced lookup compaction will be performed to flush L0
289293
/// files to higher level. This option is only valid when lookup-compact mode is gentle. No

include/paimon/file_store_commit.h

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -135,6 +135,17 @@ class PAIMON_EXPORT FileStoreCommit {
135135
virtual Status DropPartition(const std::vector<std::map<std::string, std::string>>& partitions,
136136
int64_t commit_identifier) = 0;
137137

138+
/// Configure row-id conflict checking from a specific snapshot id.
139+
///
140+
/// If set to a snapshot id, commit conflict detection will additionally validate row-id
141+
/// conflicts against snapshots after that id. Passing std::nullopt disables this behavior.
142+
///
143+
/// @param row_id_check_from_snapshot Snapshot id to start row-id conflict checks from, or
144+
/// std::nullopt to disable.
145+
/// @return Current commit object for chaining.
146+
virtual FileStoreCommit& RowIdCheckConflict(
147+
std::optional<int64_t> row_id_check_from_snapshot) = 0;
148+
138149
/// Retrieve metrics related to commit operations.
139150
///
140151
/// @return A shared pointer to a `Metrics` object containing commit metrics.

src/paimon/CMakeLists.txt

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -290,6 +290,7 @@ set(PAIMON_CORE_SRCS
290290
core/operation/commit/conflict_detection.cpp
291291
core/operation/commit/commit_scanner.cpp
292292
core/operation/commit/commit_changes_provider.cpp
293+
core/operation/commit/compacted_changelog_path_resolver.cpp
293294
core/operation/commit/overwrite_changes_provider.cpp
294295
core/operation/commit/row_id_column_conflict_checker.cpp
295296
core/operation/commit/manifest_entry_changes.cpp
@@ -704,8 +705,10 @@ if(PAIMON_BUILD_TESTS)
704705
core/operation/metrics/compaction_metrics_test.cpp
705706
core/operation/data_evolution_file_store_scan_test.cpp
706707
core/operation/data_evolution_split_read_test.cpp
708+
core/operation/commit/compacted_changelog_path_resolver_test.cpp
707709
core/operation/commit/conflict_detection_test.cpp
708710
core/operation/commit/manifest_entry_changes_test.cpp
711+
core/operation/commit/row_id_column_conflict_checker_test.cpp
709712
core/operation/commit/row_tracking_commit_utils_test.cpp
710713
core/operation/commit/retry_waiter_test.cpp
711714
core/operation/key_value_file_store_write_test.cpp

src/paimon/common/defs.cpp

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -142,6 +142,7 @@ const char Options::NUM_LEVELS[] = "num-levels";
142142
const char Options::COMPACTION_FORCE_UP_LEVEL_0[] = "compaction.force-up-level-0";
143143
const char Options::OVERWRITE_UPGRADE[] = "overwrite-upgrade";
144144
const char Options::DYNAMIC_PARTITION_OVERWRITE[] = "dynamic-partition-overwrite";
145+
const char Options::PK_CLUSTERING_OVERRIDE[] = "pk-clustering-override";
145146
const char Options::LOOKUP_WAIT[] = "lookup-wait";
146147
const char Options::LOOKUP_COMPACT[] = "lookup-compact";
147148
const char Options::LOOKUP_COMPACT_MAX_INTERVAL[] = "lookup-compact.max-interval";

src/paimon/core/core_options.cpp

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -453,6 +453,7 @@ struct CoreOptions::Impl {
453453
bool commit_discard_duplicate_files = false;
454454
bool dynamic_partition_overwrite = true;
455455
bool overwrite_upgrade = true;
456+
bool pk_clustering_override = false;
456457
bool compaction_force_rewrite_all_files = false;
457458
bool compaction_force_up_level_0 = false;
458459
std::optional<std::string> global_index_external_path;
@@ -664,6 +665,13 @@ struct CoreOptions::Impl {
664665
// Parse overwrite-upgrade - whether to try upgrading data files after overwrite on
665666
// primary key table
666667
PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::OVERWRITE_UPGRADE, &overwrite_upgrade));
668+
// Parse pk-clustering-override - currently unsupported in paimon-cpp
669+
PAIMON_RETURN_NOT_OK(
670+
parser.Parse<bool>(Options::PK_CLUSTERING_OVERRIDE, &pk_clustering_override));
671+
if (pk_clustering_override) {
672+
return Status::Invalid(
673+
"Unsupported option: pk-clustering-override=true is not supported in paimon-cpp");
674+
}
667675
return Status::OK();
668676
}
669677

@@ -1097,6 +1105,10 @@ bool CoreOptions::OverwriteUpgrade() const {
10971105
return impl_->overwrite_upgrade;
10981106
}
10991107

1108+
bool CoreOptions::PkClusteringOverride() const {
1109+
return impl_->pk_clustering_override;
1110+
}
1111+
11001112
int32_t CoreOptions::GetCompactionMinFileNum() const {
11011113
return impl_->compaction_min_file_num;
11021114
}

src/paimon/core/core_options.h

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -114,6 +114,7 @@ class PAIMON_EXPORT CoreOptions {
114114
bool CommitDiscardDuplicateFiles() const;
115115
bool DynamicPartitionOverwrite() const;
116116
bool OverwriteUpgrade() const;
117+
bool PkClusteringOverride() const;
117118
int32_t GetCompactionMinFileNum() const;
118119
int32_t GetCompactionMaxSizeAmplificationPercent() const;
119120
int32_t GetCompactionSizeRatio() const;

src/paimon/core/core_options_test.cpp

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -69,6 +69,7 @@ TEST(CoreOptionsTest, TestDefaultValue) {
6969
ASSERT_FALSE(core_options.CommitForceCompact());
7070
ASSERT_TRUE(core_options.DynamicPartitionOverwrite());
7171
ASSERT_TRUE(core_options.OverwriteUpgrade());
72+
ASSERT_FALSE(core_options.PkClusteringOverride());
7273
ASSERT_EQ(std::numeric_limits<int64_t>::max(), core_options.GetCommitTimeout());
7374
ASSERT_EQ(10, core_options.GetCommitMaxRetries());
7475
ASSERT_EQ(10, core_options.GetCommitMinRetryWait());
@@ -318,6 +319,7 @@ TEST(CoreOptionsTest, TestFromMap) {
318319
ASSERT_TRUE(core_options.CommitForceCompact());
319320
ASSERT_FALSE(core_options.DynamicPartitionOverwrite());
320321
ASSERT_FALSE(core_options.OverwriteUpgrade());
322+
ASSERT_FALSE(core_options.PkClusteringOverride());
321323
ASSERT_EQ(120 * 1000, core_options.GetCommitTimeout());
322324
ASSERT_EQ(20, core_options.GetCommitMaxRetries());
323325
ASSERT_EQ(5, core_options.GetCommitMinRetryWait());
@@ -492,6 +494,24 @@ TEST(CoreOptionsTest, TestDynamicPartitionOverwriteOption) {
492494
}
493495
}
494496

497+
TEST(CoreOptionsTest, TestPkClusteringOverrideOption) {
498+
{
499+
ASSERT_OK_AND_ASSIGN(CoreOptions core_options, CoreOptions::FromMap({}));
500+
ASSERT_FALSE(core_options.PkClusteringOverride());
501+
}
502+
503+
{
504+
ASSERT_OK_AND_ASSIGN(
505+
CoreOptions core_options,
506+
CoreOptions::FromMap({{Options::PK_CLUSTERING_OVERRIDE, "false"}}));
507+
ASSERT_FALSE(core_options.PkClusteringOverride());
508+
}
509+
510+
ASSERT_NOK_WITH_MSG(
511+
CoreOptions::FromMap({{Options::PK_CLUSTERING_OVERRIDE, "true"}}),
512+
"Unsupported option: pk-clustering-override=true is not supported in paimon-cpp");
513+
}
514+
495515
TEST(CoreOptionsTest, TestNumSortedRunsStopTriggerFloorAndDefault) {
496516
{
497517
std::map<std::string, std::string> options = {

src/paimon/core/operation/commit/commit_changes_provider.cpp

Lines changed: 33 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -20,20 +20,41 @@
2020

2121
namespace paimon {
2222

23-
StaticCommitChangesProvider::StaticCommitChangesProvider(
23+
namespace {
24+
25+
class FixedInputCommitChangesProvider final : public CommitChangesProvider {
26+
public:
27+
FixedInputCommitChangesProvider(const std::vector<ManifestEntry>& delta_files,
28+
const std::vector<ManifestEntry>& changelog_files,
29+
const std::vector<IndexManifestEntry>& index_entries)
30+
: delta_files_(delta_files),
31+
changelog_files_(changelog_files),
32+
index_entries_(index_entries) {}
33+
34+
Status Provide(const std::optional<Snapshot>&,
35+
std::vector<ManifestEntry>* delta_files,
36+
std::vector<ManifestEntry>* changelog_files,
37+
std::vector<IndexManifestEntry>* index_entries) const override {
38+
*delta_files = delta_files_;
39+
*changelog_files = changelog_files_;
40+
*index_entries = index_entries_;
41+
return Status::OK();
42+
}
43+
44+
private:
45+
const std::vector<ManifestEntry>& delta_files_;
46+
const std::vector<ManifestEntry>& changelog_files_;
47+
const std::vector<IndexManifestEntry>& index_entries_;
48+
};
49+
50+
} // namespace
51+
52+
std::shared_ptr<CommitChangesProvider> CommitChangesProvider::Provider(
2453
const std::vector<ManifestEntry>& delta_files,
2554
const std::vector<ManifestEntry>& changelog_files,
26-
const std::vector<IndexManifestEntry>& index_entries)
27-
: delta_files_(delta_files), changelog_files_(changelog_files), index_entries_(index_entries) {}
28-
29-
Status StaticCommitChangesProvider::Provide(const std::optional<Snapshot>&,
30-
std::vector<ManifestEntry>* delta_files,
31-
std::vector<ManifestEntry>* changelog_files,
32-
std::vector<IndexManifestEntry>* index_entries) const {
33-
*delta_files = delta_files_;
34-
*changelog_files = changelog_files_;
35-
*index_entries = index_entries_;
36-
return Status::OK();
55+
const std::vector<IndexManifestEntry>& index_entries) {
56+
return std::make_shared<FixedInputCommitChangesProvider>(delta_files, changelog_files,
57+
index_entries);
3758
}
3859

3960
} // namespace paimon

src/paimon/core/operation/commit/commit_changes_provider.h

Lines changed: 6 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717
#pragma once
1818

1919
#include <functional>
20+
#include <memory>
2021
#include <optional>
2122
#include <vector>
2223

@@ -32,27 +33,15 @@ class CommitChangesProvider {
3233
public:
3334
virtual ~CommitChangesProvider() = default;
3435

36+
static std::shared_ptr<CommitChangesProvider> Provider(
37+
const std::vector<ManifestEntry>& delta_files,
38+
const std::vector<ManifestEntry>& changelog_files,
39+
const std::vector<IndexManifestEntry>& index_entries);
40+
3541
virtual Status Provide(const std::optional<Snapshot>& latest_snapshot,
3642
std::vector<ManifestEntry>* delta_files,
3743
std::vector<ManifestEntry>* changelog_files,
3844
std::vector<IndexManifestEntry>* index_entries) const = 0;
3945
};
4046

41-
class StaticCommitChangesProvider final : public CommitChangesProvider {
42-
public:
43-
StaticCommitChangesProvider(const std::vector<ManifestEntry>& delta_files,
44-
const std::vector<ManifestEntry>& changelog_files,
45-
const std::vector<IndexManifestEntry>& index_entries);
46-
47-
Status Provide(const std::optional<Snapshot>& latest_snapshot,
48-
std::vector<ManifestEntry>* delta_files,
49-
std::vector<ManifestEntry>* changelog_files,
50-
std::vector<IndexManifestEntry>* index_entries) const override;
51-
52-
private:
53-
const std::vector<ManifestEntry>& delta_files_;
54-
const std::vector<ManifestEntry>& changelog_files_;
55-
const std::vector<IndexManifestEntry>& index_entries_;
56-
};
57-
5847
} // namespace paimon
Lines changed: 79 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,79 @@
1+
/*
2+
* Copyright 2024-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/operation/commit/compacted_changelog_path_resolver.h"
18+
19+
#include <string>
20+
#include <vector>
21+
22+
#include "paimon/common/utils/path_util.h"
23+
#include "paimon/common/utils/string_utils.h"
24+
25+
namespace paimon {
26+
27+
bool CompactedChangelogPathResolver::IsCompactedChangelogPath(const std::string& path) {
28+
const std::string file_name = PathUtil::GetName(path);
29+
return StringUtils::StartsWith(file_name, "compacted-changelog-");
30+
}
31+
32+
std::string CompactedChangelogPathResolver::Resolve(const std::string& path) {
33+
if (!IsCompactedChangelogPath(path)) {
34+
return path;
35+
}
36+
const std::string file_name = PathUtil::GetName(path);
37+
38+
const size_t dot_pos = file_name.find_last_of('.');
39+
if (dot_pos == std::string::npos || dot_pos + 1 >= file_name.size()) {
40+
return path;
41+
}
42+
43+
const std::string name_without_ext = file_name.substr(0, dot_pos);
44+
const std::string format = file_name.substr(dot_pos + 1);
45+
const size_t dollar_pos = name_without_ext.find('$');
46+
if (dollar_pos == std::string::npos || dollar_pos + 1 >= name_without_ext.size()) {
47+
return path;
48+
}
49+
50+
const std::string base_name = name_without_ext.substr(0, dollar_pos);
51+
const std::string suffix = name_without_ext.substr(dollar_pos + 1);
52+
const std::vector<std::string> split_tokens = StringUtils::Split(suffix, "-", false);
53+
54+
// Real compacted changelog path pattern: ...$bucket-len.ext
55+
if (split_tokens.size() == 2) {
56+
return path;
57+
}
58+
59+
// Fake compacted changelog path pattern: ...$bucket-len-offset-sliceLen.ext
60+
if (split_tokens.size() < 4) {
61+
return path;
62+
}
63+
64+
const std::string& bucket = split_tokens[0];
65+
const std::string& total_len = split_tokens[1];
66+
const std::string real_file_name =
67+
base_name + "$" + bucket + "-" + total_len + "." + format;
68+
69+
const std::string parent = PathUtil::GetParentDirPath(path);
70+
const std::string grand_parent = PathUtil::GetParentDirPath(parent);
71+
if (parent.empty() || grand_parent.empty()) {
72+
return path;
73+
}
74+
75+
return PathUtil::JoinPath(PathUtil::JoinPath(grand_parent, "bucket-" + bucket),
76+
real_file_name);
77+
}
78+
79+
} // namespace paimon

0 commit comments

Comments
 (0)