wangyong9999 commented on code in PR #245:
URL: https://github.com/apache/paimon-cpp/pull/245#discussion_r4008111584
##########
include/paimon/defs.h:
##########
@@ -396,6 +396,10 @@ struct PAIMON_EXPORT Options {
/// a table is never compacted.
static const char DELETION_VECTORS_ENABLED[];
+ /// "pk-clustering-override" - Whether primary-key clustering columns
override the primary
Review Comment:
Updated the description to match Java and added
`CoreOptions::PkClusteringOverrideEnabled()`. Schema validation now uses the
accessor.
##########
src/paimon/core/schema/schema_validation.cpp:
##########
@@ -137,6 +144,58 @@ Status ValidatePerLevelOption(
return Status::OK();
}
+std::vector<std::string> PrimaryKeyBTreeIndexColumns(
+ const std::map<std::string, std::string>& options) {
+ auto iter = options.find(Options::PK_BTREE_INDEX_COLUMNS);
Review Comment:
Moved this into `CoreOptions::GetPrimaryKeyBTreeIndexColumns()` and added
parsing coverage.
##########
include/paimon/defs.h:
##########
@@ -396,6 +396,10 @@ struct PAIMON_EXPORT Options {
/// a table is never compacted.
static const char DELETION_VECTORS_ENABLED[];
+ /// "pk-clustering-override" - Whether primary-key clustering columns
override the primary
+ /// keys when clustering data. Default value is false.
+ static const char PK_CLUSTERING_OVERRIDE[];
Review Comment:
The Java condition allows first-row + deletion vectors when clustering
override is enabled. C++ still rejects that mode in the commit path, so I kept
the restriction rather than enabling an unsupported combination.
##########
src/paimon/core/index/pk/bucketed_primary_key_index_maintainer.cpp:
##########
@@ -0,0 +1,294 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "paimon/core/index/pk/bucketed_primary_key_index_maintainer.h"
+
+#include <algorithm>
+#include <map>
+#include <set>
+#include <unordered_set>
+#include <utility>
+
+#include "fmt/format.h"
+#include "paimon/common/global_index/btree/btree_defs.h"
+#include "paimon/core/index/global_index_meta.h"
+#include "paimon/core/index/index_file_handler.h"
+#include "paimon/core/index/pk/primary_key_index_source_file.h"
+#include "paimon/core/index/pk/primary_key_index_source_policy.h"
+#include "paimon/core/index/pksorted/pk_sorted_bucket_index_state.h"
+#include "paimon/core/index/pksorted/pk_sorted_index_builder.h"
+#include "paimon/core/index/pksorted/pk_sorted_index_group.h"
+#include "paimon/core/utils/commit_increment.h"
+#include "paimon/logging.h"
+
+namespace paimon {
+namespace {
+
+Logger* GetLogger() {
+ static std::unique_ptr<Logger> logger =
Logger::GetLogger("BucketedPrimaryKeyIndexMaintainer");
+ return logger.get();
+}
+
+void RemoveDataFiles(const std::vector<std::shared_ptr<DataFileMeta>>& files,
+ std::map<std::string, std::shared_ptr<DataFileMeta>>*
active) {
+ for (const std::shared_ptr<DataFileMeta>& file : files) {
+ if (file != nullptr) {
+ active->erase(file->file_name);
+ }
+ }
+}
+
+Status AddSourceFiles(const std::vector<std::shared_ptr<DataFileMeta>>& files,
+ std::map<std::string, std::shared_ptr<DataFileMeta>>*
active) {
+ for (const std::shared_ptr<DataFileMeta>& file : files) {
+ if (file == nullptr) {
+ return Status::Invalid("Primary-key index data increment contains
a null file.");
+ }
+ if (PrimaryKeyIndexSourcePolicy::ShouldRead(*file)) {
+ (*active)[file->file_name] = file;
+ }
+ }
+ return Status::OK();
+}
+
+Status ValidateAppendFiles(const std::vector<std::shared_ptr<DataFileMeta>>&
files) {
+ for (const std::shared_ptr<DataFileMeta>& file : files) {
+ if (file == nullptr) {
+ return Status::Invalid("Primary-key index append increment
contains a null file.");
+ }
+ if (PrimaryKeyIndexSourcePolicy::ShouldRead(*file)) {
+ return Status::Invalid(fmt::format(
+ "Append file {} must not be a primary-key sorted-index
source.", file->file_name));
+ }
+ }
+ return Status::OK();
+}
+
+std::string PayloadIdentity(const std::shared_ptr<IndexFileMeta>& payload) {
+ if (payload == nullptr) {
+ return std::string();
+ }
+ return payload->ExternalPath().value_or(payload->FileName());
+}
+
+void AddUniquePayload(const std::shared_ptr<IndexFileMeta>& payload,
+ std::unordered_set<std::string>* identities,
+ std::vector<std::shared_ptr<IndexFileMeta>>* payloads) {
+ std::string identity = PayloadIdentity(payload);
+ if (!identity.empty() && identities->insert(identity).second) {
+ payloads->push_back(payload);
+ }
+}
+
+bool IsPrimaryKeyBTreePayload(const std::shared_ptr<IndexFileMeta>& payload) {
+ return payload != nullptr && payload->IndexType() ==
BtreeDefs::kIdentifier &&
+ IndexFileHandler::IsPrimaryKeySourceIndex(*payload);
+}
+
+bool CoversAllSources(const std::vector<PrimaryKeyIndexSourceFile>&
group_sources,
+ const std::vector<PrimaryKeyIndexSourceFile>&
desired_sources) {
+ size_t group_index = 0;
+ for (const PrimaryKeyIndexSourceFile& desired : desired_sources) {
+ while (group_index < group_sources.size() &&
+ group_sources[group_index].file_name < desired.file_name) {
+ group_index++;
+ }
+ if (group_index == group_sources.size() || group_sources[group_index]
!= desired) {
+ return false;
+ }
+ group_index++;
+ }
+ return true;
+}
+
+} // namespace
+
+Result<std::shared_ptr<BucketedPrimaryKeyIndexMaintainer::Factory>>
Review Comment:
Removed `Result` from `Factory::Create()` and updated its callers.
`CreateMaintainer()` remains fallible.
##########
src/paimon/core/index/pk/bucketed_primary_key_index_maintainer.cpp:
##########
@@ -0,0 +1,294 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "paimon/core/index/pk/bucketed_primary_key_index_maintainer.h"
+
+#include <algorithm>
+#include <map>
+#include <set>
+#include <unordered_set>
+#include <utility>
+
+#include "fmt/format.h"
+#include "paimon/common/global_index/btree/btree_defs.h"
+#include "paimon/core/index/global_index_meta.h"
+#include "paimon/core/index/index_file_handler.h"
+#include "paimon/core/index/pk/primary_key_index_source_file.h"
+#include "paimon/core/index/pk/primary_key_index_source_policy.h"
+#include "paimon/core/index/pksorted/pk_sorted_bucket_index_state.h"
+#include "paimon/core/index/pksorted/pk_sorted_index_builder.h"
+#include "paimon/core/index/pksorted/pk_sorted_index_group.h"
+#include "paimon/core/utils/commit_increment.h"
+#include "paimon/logging.h"
+
+namespace paimon {
+namespace {
+
+Logger* GetLogger() {
+ static std::unique_ptr<Logger> logger =
Logger::GetLogger("BucketedPrimaryKeyIndexMaintainer");
+ return logger.get();
+}
+
+void RemoveDataFiles(const std::vector<std::shared_ptr<DataFileMeta>>& files,
+ std::map<std::string, std::shared_ptr<DataFileMeta>>*
active) {
+ for (const std::shared_ptr<DataFileMeta>& file : files) {
+ if (file != nullptr) {
+ active->erase(file->file_name);
+ }
+ }
+}
+
+Status AddSourceFiles(const std::vector<std::shared_ptr<DataFileMeta>>& files,
+ std::map<std::string, std::shared_ptr<DataFileMeta>>*
active) {
+ for (const std::shared_ptr<DataFileMeta>& file : files) {
+ if (file == nullptr) {
+ return Status::Invalid("Primary-key index data increment contains
a null file.");
+ }
+ if (PrimaryKeyIndexSourcePolicy::ShouldRead(*file)) {
+ (*active)[file->file_name] = file;
+ }
+ }
+ return Status::OK();
+}
+
+Status ValidateAppendFiles(const std::vector<std::shared_ptr<DataFileMeta>>&
files) {
+ for (const std::shared_ptr<DataFileMeta>& file : files) {
+ if (file == nullptr) {
+ return Status::Invalid("Primary-key index append increment
contains a null file.");
+ }
+ if (PrimaryKeyIndexSourcePolicy::ShouldRead(*file)) {
+ return Status::Invalid(fmt::format(
+ "Append file {} must not be a primary-key sorted-index
source.", file->file_name));
+ }
+ }
+ return Status::OK();
+}
+
+std::string PayloadIdentity(const std::shared_ptr<IndexFileMeta>& payload) {
+ if (payload == nullptr) {
+ return std::string();
+ }
+ return payload->ExternalPath().value_or(payload->FileName());
+}
+
+void AddUniquePayload(const std::shared_ptr<IndexFileMeta>& payload,
+ std::unordered_set<std::string>* identities,
+ std::vector<std::shared_ptr<IndexFileMeta>>* payloads) {
+ std::string identity = PayloadIdentity(payload);
+ if (!identity.empty() && identities->insert(identity).second) {
+ payloads->push_back(payload);
+ }
+}
+
+bool IsPrimaryKeyBTreePayload(const std::shared_ptr<IndexFileMeta>& payload) {
+ return payload != nullptr && payload->IndexType() ==
BtreeDefs::kIdentifier &&
+ IndexFileHandler::IsPrimaryKeySourceIndex(*payload);
+}
+
+bool CoversAllSources(const std::vector<PrimaryKeyIndexSourceFile>&
group_sources,
+ const std::vector<PrimaryKeyIndexSourceFile>&
desired_sources) {
+ size_t group_index = 0;
+ for (const PrimaryKeyIndexSourceFile& desired : desired_sources) {
+ while (group_index < group_sources.size() &&
+ group_sources[group_index].file_name < desired.file_name) {
+ group_index++;
+ }
+ if (group_index == group_sources.size() || group_sources[group_index]
!= desired) {
+ return false;
+ }
+ group_index++;
+ }
+ return true;
+}
+
+} // namespace
+
+Result<std::shared_ptr<BucketedPrimaryKeyIndexMaintainer::Factory>>
+BucketedPrimaryKeyIndexMaintainer::Factory::Create(
+ const std::string& root_path, const std::string& branch,
+ const std::shared_ptr<TableSchema>& table_schema,
+ const std::vector<PrimaryKeyIndexDefinition>& definitions,
+ const std::shared_ptr<FileStorePathFactory>& path_factory,
+ const std::shared_ptr<IndexFileHandler>& index_file_handler, const
CoreOptions& options,
+ const std::shared_ptr<IOManager>& io_manager, bool
enable_multi_thread_spill,
+ const std::shared_ptr<Executor>& executor, const
std::shared_ptr<MemoryPool>& pool) {
+ std::vector<PrimaryKeyIndexDefinition> btree_definitions;
+ for (const PrimaryKeyIndexDefinition& definition : definitions) {
+ if (definition.GetFamily() ==
PrimaryKeyIndexDefinition::Family::BTREE) {
+ btree_definitions.push_back(definition);
+ }
+ }
+ std::sort(btree_definitions.begin(), btree_definitions.end(),
+ [](const PrimaryKeyIndexDefinition& left, const
PrimaryKeyIndexDefinition& right) {
+ return left.FieldId() < right.FieldId();
+ });
+ return std::shared_ptr<Factory>(new Factory(
+ root_path, branch, table_schema, std::move(btree_definitions),
path_factory,
+ index_file_handler, options, io_manager, enable_multi_thread_spill,
executor, pool));
+}
+
+Result<std::shared_ptr<BucketedPrimaryKeyIndexMaintainer>>
+BucketedPrimaryKeyIndexMaintainer::Factory::CreateMaintainer(
+ const BinaryRow& partition, int32_t bucket,
+ const std::vector<std::shared_ptr<DataFileMeta>>& restored_data_files,
+ const std::vector<std::shared_ptr<IndexFileMeta>>& restored_payloads)
const {
+ std::map<std::string, std::shared_ptr<DataFileMeta>> active_data_files;
+ PAIMON_RETURN_NOT_OK(AddSourceFiles(restored_data_files,
&active_data_files));
+ std::vector<FieldMaintainer> fields;
+ fields.reserve(definitions_.size());
+ for (const PrimaryKeyIndexDefinition& definition : definitions_) {
+ PAIMON_ASSIGN_OR_RAISE(
+ std::unique_ptr<PkSortedIndexBuilder> builder,
+ PkSortedIndexBuilder::Create(root_path_, branch_, partition,
bucket, table_schema_,
+ definition, path_factory_, options_,
io_manager_,
+ enable_multi_thread_spill_,
executor_, pool_));
+ fields.push_back(
+ FieldMaintainer{definition,
std::shared_ptr<PkSortedIndexBuilder>(std::move(builder))});
+ }
+ return std::shared_ptr<BucketedPrimaryKeyIndexMaintainer>(new
BucketedPrimaryKeyIndexMaintainer(
+ std::move(fields), std::move(active_data_files), restored_payloads));
+}
+
+Status BucketedPrimaryKeyIndexMaintainer::PrepareCommit(CommitIncrement*
increment) {
+ if (increment == nullptr) {
+ return Status::Invalid("Primary-key index commit increment is null.");
+ }
+ auto previous_data_files = active_data_files_;
+ const DataIncrement& data_increment = increment->GetNewFilesIncrement();
+ const CompactIncrement& compact_increment =
increment->GetCompactIncrement();
+ PAIMON_RETURN_NOT_OK(ValidateAppendFiles(data_increment.NewFiles()));
+ RemoveDataFiles(compact_increment.CompactBefore(), &active_data_files_);
+ Status update_status = AddSourceFiles(compact_increment.CompactAfter(),
&active_data_files_);
+ if (!update_status.ok()) {
+ active_data_files_ = std::move(previous_data_files);
Review Comment:
Changes now go into a temporary copy, assigned back only after validation
succeeds. Added a regression test verifying that an invalid increment leaves
the original state unchanged.
--
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]
