gavinchou commented on code in PR #65859:
URL: https://github.com/apache/doris/pull/65859#discussion_r3657671815
##########
cloud/src/meta-service/meta_service_partition.cpp:
##########
@@ -83,6 +86,100 @@ static TxnErrorCode check_recycle_key_exist(Transaction*
txn, const std::string&
return txn->get(key, &val);
}
+static bool is_table_stream_versioned_write(MultiVersionStatus
multi_version_status) {
+ return multi_version_status ==
MultiVersionStatus::MULTI_VERSION_WRITE_ONLY ||
+ multi_version_status ==
MultiVersionStatus::MULTI_VERSION_READ_WRITE;
+}
+
+static bool validate_table_stream_index_request(const IndexRequest* request,
MetaServiceCode& code,
+ std::string& msg) {
+ if (request->index_ids_size() != 1 || !request->has_db_id() ||
!request->has_stream_db_id()) {
+ code = MetaServiceCode::INVALID_ARGUMENT;
+ msg = "table stream requires exactly one index_id, db_id and
stream_db_id";
+ return false;
+ }
+ if ((request->has_is_new_table() && request->is_new_table()) ||
+ !request->partition_ids().empty()) {
+ code = MetaServiceCode::INVALID_ARGUMENT;
+ msg = "table stream index must not create a table or physical
partitions";
+ return false;
+ }
+ return true;
+}
+
+static bool table_stream_recycle_index_matches(const RecycleIndexPB&
recycle_index,
+ const IndexRequest& request) {
+ return recycle_index.state() == RecycleIndexPB::PREPARED &&
+ recycle_index.object_type() == IndexObjectTypePB::TABLE_STREAM &&
+ recycle_index.has_db_id() && recycle_index.db_id() ==
request.db_id() &&
+ recycle_index.table_id() == request.table_id() &&
recycle_index.has_stream_db_id() &&
+ recycle_index.stream_db_id() == request.stream_db_id();
+}
+
+static bool table_stream_recycle_index_matches(const RecycleIndexPB&
recycle_index,
+ const PartitionRequest&
request) {
+ return recycle_index.state() == RecycleIndexPB::PREPARED &&
+ recycle_index.object_type() == IndexObjectTypePB::TABLE_STREAM &&
+ recycle_index.has_db_id() && recycle_index.db_id() ==
request.db_id() &&
+ recycle_index.table_id() == request.table_id() &&
recycle_index.has_stream_db_id() &&
+ recycle_index.stream_db_id() == request.stream_db_id();
+}
+
+static TxnErrorCode table_stream_offset_exists(Transaction* txn, const
std::string& instance_id,
+ int64_t base_db_id, int64_t
base_table_id,
+ int64_t stream_db_id, int64_t
stream_id) {
+ const std::string begin = table_stream_offset_key_prefix(instance_id,
base_db_id, base_table_id,
+ stream_db_id,
stream_id);
+ std::unique_ptr<RangeGetIterator> it;
+ TxnErrorCode err = txn->get(begin, lexical_end(begin), &it, false, 1);
+ if (err != TxnErrorCode::TXN_OK) {
+ return err;
+ }
+ return it->has_next() ? TxnErrorCode::TXN_OK :
TxnErrorCode::TXN_KEY_NOT_FOUND;
Review Comment:
what if `it->more()` ?
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
