You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by w4...@apache.org on 2025/12/22 10:47:13 UTC
(doris) branch master updated: [chore](partition) Write versioned partition key when commit index (#59078)
This is an automated email from the ASF dual-hosted git repository.
w41ter pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 3f2a7128135 [chore](partition) Write versioned partition key when commit index (#59078)
3f2a7128135 is described below
commit 3f2a71281353d28be81c6b60c019132ed1b8b919
Author: Yixuan Wang <wa...@selectdb.com>
AuthorDate: Mon Dec 22 18:47:04 2025 +0800
[chore](partition) Write versioned partition key when commit index (#59078)
---
cloud/src/meta-service/meta_service_partition.cpp | 31 ++++++++++++++++++
cloud/test/meta_service_operation_log_test.cpp | 36 ++++++++++++++++++++-
.../org/apache/doris/alter/CloudRollupJobV2.java | 2 +-
.../apache/doris/alter/CloudSchemaChangeJobV2.java | 2 +-
.../doris/clone/DynamicPartitionScheduler.java | 2 +-
.../cloud/datasource/CloudInternalCatalog.java | 37 +++++++++++++++++-----
.../apache/doris/datasource/InternalCatalog.java | 26 +++++++++++----
gensrc/proto/cloud.proto | 2 ++
8 files changed, 119 insertions(+), 19 deletions(-)
diff --git a/cloud/src/meta-service/meta_service_partition.cpp b/cloud/src/meta-service/meta_service_partition.cpp
index e305726984d..811fc219b99 100644
--- a/cloud/src/meta-service/meta_service_partition.cpp
+++ b/cloud/src/meta-service/meta_service_partition.cpp
@@ -286,6 +286,37 @@ void MetaServiceImpl::commit_index(::google::protobuf::RpcController* controller
}
}
+ // Save the partition meta keys
+ if (is_versioned_write) {
+ for (auto partition_id : request->partition_ids()) {
+ int64_t db_id = request->db_id();
+ int64_t table_id = request->table_id();
+ std::string part_meta_key = versioned::meta_partition_key({instance_id, partition_id});
+ std::string part_index_key =
+ versioned::partition_index_key({instance_id, partition_id});
+ std::string part_inverted_index_key = versioned::partition_inverted_index_key(
+ {instance_id, db_id, table_id, partition_id});
+ PartitionIndexPB part_index_pb;
+ part_index_pb.set_db_id(db_id);
+ part_index_pb.set_table_id(table_id);
+ std::string part_index_value;
+ if (!part_index_pb.SerializeToString(&part_index_value)) {
+ code = MetaServiceCode::PROTOBUF_SERIALIZE_ERR;
+ msg = fmt::format("failed to serialize PartitionIndexPB");
+ LOG_WARNING(msg).tag("part_id", partition_id);
+ return;
+ }
+ versioned_put(txn.get(), part_meta_key, "");
+ txn->put(part_inverted_index_key, "");
+ txn->put(part_index_key, part_index_value);
+
+ LOG(INFO) << "xxx put versioned partition index key=" << hex(part_index_key)
+ << " partition_id=" << partition_id;
+
+ commit_index_log.add_partition_ids(partition_id);
+ }
+ }
+
if (request->has_is_new_table() && request->is_new_table()) {
if (is_versioned_read) {
// Read the table version, to build the operation log visible version range.
diff --git a/cloud/test/meta_service_operation_log_test.cpp b/cloud/test/meta_service_operation_log_test.cpp
index af2da9b23f9..5a5c46a7603 100644
--- a/cloud/test/meta_service_operation_log_test.cpp
+++ b/cloud/test/meta_service_operation_log_test.cpp
@@ -425,6 +425,7 @@ TEST(MetaServiceOperationLogTest, CommitIndexLog) {
constexpr int64_t db_id = 123;
constexpr int64_t table_id = 10001;
constexpr int64_t index_id = 10002;
+ constexpr int64_t part_id = 10003;
{
// write instance
@@ -461,12 +462,17 @@ TEST(MetaServiceOperationLogTest, CommitIndexLog) {
req.set_table_id(table_id);
req.add_index_ids(index_id);
req.set_is_new_table(true);
+ for (size_t i = 0; i < 5; i++) {
+ req.add_partition_ids(part_id + i);
+ }
meta_service->commit_index(&ctrl, &req, &res, nullptr);
ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << res.status().DebugString();
}
auto txn_kv = meta_service->txn_kv();
Versionstamp version1;
+ Versionstamp version2;
+
{
// Verify index meta/index/inverted indexes are exists
std::string index_meta_key = versioned::meta_index_key({instance_id, index_id});
@@ -488,7 +494,35 @@ TEST(MetaServiceOperationLogTest, CommitIndexLog) {
ASSERT_EQ(index_index.table_id(), table_id);
}
- Versionstamp version2;
+ {
+ // Verify table version exists
+ MetaReader meta_reader(instance_id, txn_kv.get());
+ ASSERT_EQ(meta_reader.get_table_version(table_id, &version2), TxnErrorCode::TXN_OK);
+ }
+
+ {
+ for (size_t i = 0; i < 5; i++) {
+ std::string part_index_key = versioned::partition_index_key({instance_id, part_id + i});
+ std::string part_meta_key = versioned::meta_partition_key({instance_id, part_id + i});
+ std::string part_inverted_index_key = versioned::partition_inverted_index_key(
+ {instance_id, db_id, table_id, part_id + i});
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+ std::string value;
+ ASSERT_EQ(versioned_get(txn.get(), part_meta_key, &version1, &value),
+ TxnErrorCode::TXN_OK);
+
+ ASSERT_EQ(txn->get(part_index_key, &value), TxnErrorCode::TXN_OK);
+
+ PartitionIndexPB part_index;
+ ASSERT_TRUE(part_index.ParseFromString(value));
+ ASSERT_EQ(part_index.db_id(), db_id);
+ ASSERT_EQ(part_index.table_id(), table_id);
+
+ ASSERT_EQ(txn->get(part_inverted_index_key, &value), TxnErrorCode::TXN_OK);
+ }
+ }
+
{
// Verify table version exists
MetaReader meta_reader(instance_id, txn_kv.get());
diff --git a/fe/fe-core/src/main/java/org/apache/doris/alter/CloudRollupJobV2.java b/fe/fe-core/src/main/java/org/apache/doris/alter/CloudRollupJobV2.java
index bae2b41bbab..5aab530f395 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/alter/CloudRollupJobV2.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/alter/CloudRollupJobV2.java
@@ -110,7 +110,7 @@ public class CloudRollupJobV2 extends RollupJobV2 {
rollupIndexList.add(rollupIndexId);
try {
((CloudInternalCatalog) Env.getCurrentInternalCatalog())
- .commitMaterializedIndex(dbId, tableId, rollupIndexList, false);
+ .commitMaterializedIndex(dbId, tableId, rollupIndexList, null, false);
} catch (Exception e) {
LOG.warn("commitMaterializedIndex Exception:{}", e);
throw new AlterCancelException(e.getMessage());
diff --git a/fe/fe-core/src/main/java/org/apache/doris/alter/CloudSchemaChangeJobV2.java b/fe/fe-core/src/main/java/org/apache/doris/alter/CloudSchemaChangeJobV2.java
index 0d02dc1bc48..f94da677ab7 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/alter/CloudSchemaChangeJobV2.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/alter/CloudSchemaChangeJobV2.java
@@ -83,7 +83,7 @@ public class CloudSchemaChangeJobV2 extends SchemaChangeJobV2 {
indexIdMap.keySet().stream().collect(Collectors.toList());
try {
((CloudInternalCatalog) Env.getCurrentInternalCatalog())
- .commitMaterializedIndex(dbId, tableId, shadowIdxList, false);
+ .commitMaterializedIndex(dbId, tableId, shadowIdxList, null, false);
} catch (Exception e) {
LOG.warn("commitMaterializedIndex exception:", e);
throw new AlterCancelException(e.getMessage());
diff --git a/fe/fe-core/src/main/java/org/apache/doris/clone/DynamicPartitionScheduler.java b/fe/fe-core/src/main/java/org/apache/doris/clone/DynamicPartitionScheduler.java
index 88a812bdc90..f77ec517667 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/clone/DynamicPartitionScheduler.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/clone/DynamicPartitionScheduler.java
@@ -844,7 +844,7 @@ public class DynamicPartitionScheduler extends MasterDaemon {
throw new Exception("debug point FE.DynamicPartitionScheduler.before.commitCloudPartition");
}
Env.getCurrentInternalCatalog().afterCreatePartitions(db.getId(), olapTable.getId(),
- succeedPartitionIds, indexIds, true);
+ succeedPartitionIds, indexIds, true /* isCreateTable */, false /* isBatchCommit */);
LOG.info("begin write edit log to add partitions in batch, "
+ "numPartitions: {}, db: {}, table: {}, tableId: {}",
partsInfo.size(), db.getFullName(), tableName, olapTable.getId());
diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/datasource/CloudInternalCatalog.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/datasource/CloudInternalCatalog.java
index e81433afba3..c981be74875 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/cloud/datasource/CloudInternalCatalog.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/datasource/CloudInternalCatalog.java
@@ -451,24 +451,39 @@ public class CloudInternalCatalog extends InternalCatalog {
}
}
+ /**
+ * Commit partition creation to MetaService.
+ *
+ * @param partitionIds Partition IDs to commit
+ * @param indexIds Index IDs to commit
+ * @param isCreateTable Whether this is part of table creation
+ * @param isBatchCommit If true, use commitMaterializedIndex (commit_index RPC)
+ * to batch commit all partitions
+ * and indexes in one MetaService call for better
+ * performance;
+ * If false, use commitPartition (commit_partition RPC) to
+ * commit partitions separately
+ * @throws DdlException If commit to MetaService fails
+ */
@Override
public void afterCreatePartitions(long dbId, long tableId, List<Long> partitionIds, List<Long> indexIds,
- boolean isCreateTable)
+ boolean isCreateTable, boolean isBatchCommit)
throws DdlException {
- if (partitionIds == null) {
- commitMaterializedIndex(dbId, tableId, indexIds, isCreateTable);
+ if (isBatchCommit) {
+ commitMaterializedIndex(dbId, tableId, indexIds, partitionIds, isCreateTable);
} else {
commitPartition(dbId, tableId, partitionIds, indexIds);
}
if (!Config.check_create_table_recycle_key_remained) {
return;
}
- checkCreatePartitions(dbId, tableId, partitionIds, indexIds);
+ checkCreatePartitions(dbId, tableId, partitionIds, indexIds, isBatchCommit);
}
- private void checkCreatePartitions(long dbId, long tableId, List<Long> partitionIds, List<Long> indexIds)
+ private void checkCreatePartitions(long dbId, long tableId, List<Long> partitionIds, List<Long> indexIds,
+ boolean isBatchCommit)
throws DdlException {
- if (partitionIds == null) {
+ if (isBatchCommit) {
checkMaterializedIndex(dbId, tableId, indexIds);
} else {
checkPartition(dbId, tableId, partitionIds);
@@ -530,10 +545,12 @@ public class CloudInternalCatalog extends InternalCatalog {
Cloud.PartitionRequest.Builder partitionRequestBuilder = Cloud.PartitionRequest.newBuilder();
partitionRequestBuilder.setCloudUniqueId(Config.cloud_unique_id);
- partitionRequestBuilder.addAllPartitionIds(partitionIds);
partitionRequestBuilder.addAllIndexIds(indexIds);
partitionRequestBuilder.setDbId(dbId);
partitionRequestBuilder.setTableId(tableId);
+ if (partitionIds != null) {
+ partitionRequestBuilder.addAllPartitionIds(partitionIds);
+ }
final Cloud.PartitionRequest partitionRequest = partitionRequestBuilder.build();
Cloud.PartitionResponse response = null;
@@ -596,7 +613,8 @@ public class CloudInternalCatalog extends InternalCatalog {
}
}
- public void commitMaterializedIndex(long dbId, long tableId, List<Long> indexIds, boolean isCreateTable)
+ public void commitMaterializedIndex(long dbId, long tableId, List<Long> indexIds, List<Long> partitionIds,
+ boolean isCreateTable)
throws DdlException {
if (Config.enable_check_compatibility_mode) {
LOG.info("skip committing materialized index in checking compatibility mode");
@@ -609,6 +627,9 @@ public class CloudInternalCatalog extends InternalCatalog {
indexRequestBuilder.setDbId(dbId);
indexRequestBuilder.setTableId(tableId);
indexRequestBuilder.setIsNewTable(isCreateTable);
+ if (partitionIds != null) {
+ indexRequestBuilder.addAllPartitionIds(partitionIds);
+ }
final Cloud.IndexRequest indexRequest = indexRequestBuilder.build();
Cloud.IndexResponse response = null;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java
index 9179ba26b39..6b777ecd730 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java
@@ -1781,7 +1781,8 @@ public class InternalCatalog implements CatalogIf<Database> {
}
if (!isCreateTable) {
- afterCreatePartitions(db.getId(), olapTable.getId(), partitionIds, indexIds, isCreateTable);
+ afterCreatePartitions(db.getId(), olapTable.getId(), partitionIds, indexIds, isCreateTable,
+ false /* isBatchCommit */);
}
if (writeEditLog) {
Env.getCurrentEnv().getEditLog().logAddPartition(info);
@@ -2205,8 +2206,18 @@ public class InternalCatalog implements CatalogIf<Database> {
throws DdlException {
}
+ /**
+ * Commit partition creation to MetaService (Cloud mode only).
+ *
+ * @param partitionIds Partition IDs to commit
+ * @param indexIds Index IDs to commit
+ * @param isCreateTable Whether this is part of table creation
+ * @param isBatchCommit If true, use commit_index RPC to batch commit all partitions and indexes in one call;
+ * If false, use commit_partition RPC to commit partitions individually
+ * @throws DdlException If commit fails
+ */
public void afterCreatePartitions(long dbId, long tableId, List<Long> partitionIds, List<Long> indexIds,
- boolean isCreateTable)
+ boolean isCreateTable, boolean isBatchCommit)
throws DdlException {
}
@@ -2953,8 +2964,8 @@ public class InternalCatalog implements CatalogIf<Database> {
idGeneratorBuffer,
binlogConfigForTask,
partitionInfo.getDataProperty(partitionId).isStorageMediumSpecified());
- afterCreatePartitions(db.getId(), olapTable.getId(), null,
- olapTable.getIndexIdList(), true);
+ afterCreatePartitions(db.getId(), olapTable.getId(), olapTable.getPartitionIds(),
+ olapTable.getIndexIdList(), true /* isCreateTable */, true /* isBatchCommit */);
olapTable.addPartition(partition);
} else if (partitionInfo.getType() == PartitionType.RANGE
|| partitionInfo.getType() == PartitionType.LIST) {
@@ -3047,8 +3058,8 @@ public class InternalCatalog implements CatalogIf<Database> {
olapTable.getPartitionInfo().getDataProperty(partition.getId())
.setStoragePolicy(partionStoragePolicy);
}
- afterCreatePartitions(db.getId(), olapTable.getId(), null,
- olapTable.getIndexIdList(), true);
+ afterCreatePartitions(db.getId(), olapTable.getId(), olapTable.getPartitionIds(),
+ olapTable.getIndexIdList(), true /* isCreateTable */, true /* isBatchCommit */);
} else {
throw new DdlException("Unsupported partition method: " + partitionInfo.getType().name());
}
@@ -3509,7 +3520,8 @@ public class InternalCatalog implements CatalogIf<Database> {
newPartitions.add(newPartition);
}
- afterCreatePartitions(db.getId(), copiedTbl.getId(), newPartitionIds, indexIds, true);
+ afterCreatePartitions(db.getId(), copiedTbl.getId(), newPartitionIds, indexIds, true /* isCreateTable */,
+ false /* isBatchCommit */);
} catch (DdlException e) {
// create partition failed, remove all newly created tablets
diff --git a/gensrc/proto/cloud.proto b/gensrc/proto/cloud.proto
index ea5441cdc4a..ba0fbd7cc53 100644
--- a/gensrc/proto/cloud.proto
+++ b/gensrc/proto/cloud.proto
@@ -770,6 +770,7 @@ message CommitIndexLogPB {
optional int64 table_id = 2;
repeated int64 index_ids = 3;
optional bool update_table_version = 4;
+ repeated int64 partition_ids = 5;
}
message DropIndexLogPB {
@@ -1341,6 +1342,7 @@ message IndexRequest {
optional int64 db_id = 5;
optional bool is_new_table = 6; // For table version
optional string request_ip = 7;
+ repeated int64 partition_ids = 8;
}
message IndexResponse {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org