github-actions[bot] commented on code in PR #65859:
URL: https://github.com/apache/doris/pull/65859#discussion_r3664688828
##########
cloud/src/meta-service/meta_service_partition.cpp:
##########
@@ -976,6 +1445,7 @@ void
MetaServiceImpl::drop_partition(::google::protobuf::RpcController* controll
drop_partition_log.set_db_id(request->db_id());
drop_partition_log.set_table_id(request->table_id());
drop_partition_log.mutable_index_ids()->CopyFrom(request->index_ids());
+
drop_partition_log.mutable_table_streams()->CopyFrom(request->table_streams());
Review Comment:
[P1] Include Stream offset history in this drop's retention boundary
This list makes the later partition recycler call `versioned_remove_all` for
every referenced Stream offset, but this drop's `min_timestamp` is still
derived only from partition/table metadata. For example, an offset at V20 can
still be needed by a snapshot at V25 after an index operation rewrites
partition metadata at V30; a drop at V40 then records V30, so `can_recycle`
searches only `[V30,V40)`, misses V25, and deletes the V20 offset. Read the
referenced offset histories when computing the boundary (or otherwise make
physical deletion wait for their older operation logs) so active
snapshots/clones cannot lose their offsets.
##########
cloud/src/meta-service/meta_service_txn.cpp:
##########
@@ -48,6 +54,280 @@ namespace doris::cloud {
static constexpr std::string_view kMetaSyncPointDummyKey =
"__meta_service_sync_point_dummy_key__";
+static bool validate_table_stream_updates(const CommitTxnRequest* request,
MetaServiceCode& code,
+ std::string& msg) {
+ if (request->has_is_2pc() && request->is_2pc()) {
+ code = MetaServiceCode::INVALID_ARGUMENT;
+ msg = "table stream consumption does not support 2PC";
+ return false;
+ }
+ if ((request->has_is_txn_load() && request->is_txn_load()) ||
+ !request->sub_txn_infos().empty()) {
+ code = MetaServiceCode::INVALID_ARGUMENT;
+ msg = "table stream consumption does not support transaction load or
sub transactions";
+ return false;
+ }
+
+ std::set<std::pair<int64_t, int64_t>> stream_partitions;
+ for (const auto& stream_update : request->table_stream_updates()) {
+ if (!stream_update.has_identity() ||
+ !is_valid_table_stream_identity(stream_update.identity()) ||
+ stream_update.partition_updates().empty()) {
+ code = MetaServiceCode::INVALID_ARGUMENT;
+ msg = "invalid table stream update";
+ return false;
+ }
+ for (const auto& partition_update : stream_update.partition_updates())
{
+ if (!partition_update.has_partition_id() ||
partition_update.partition_id() <= 0 ||
Review Comment:
[P2] Reject negative offset TSOs before they are persisted
This validates only that `next_offset_tso` is present. For an absent offset,
`expected_state=UNKNOWN` bypasses the backward check, and any negative value
also passes the later `next <= visible_tso` check, so the commit stores a
`CONSUMED` offset such as `-1`. FE then treats it as a real consumed start
boundary and sends it to the row-binlog scan, which can produce an invalid
range or replay data. Enforce the valid TSO lower bound here (and on
create/expected offsets) before writing either projection.
##########
cloud/src/meta-service/meta_service_partition.cpp:
##########
@@ -83,6 +88,121 @@ 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()) {
Review Comment:
[P2] Reject Stream identities that cannot be read back
These checks require the identity fields to be present but never require
positive IDs. A request with `stream_id=0` (or another nonpositive binding ID)
can therefore prepare and commit offset keys, while
`is_valid_table_stream_identity` rejects the same identity in the read,
consume, and drop paths. That leaves committed metadata with no usable
lifecycle. Apply the same positive-ID validator before any Stream
prepare/partition write.
##########
cloud/src/recycler/checker.cpp:
##########
@@ -2275,6 +2333,179 @@ int InstanceChecker::scan_and_handle_kv(
return ret;
}
+// The check validates Offset values and the Latest/Versioned projection
invariant. FE Catalog is
+// the authority for Stream existence and binding, so no MS-side Stream
Mapping is checked here.
+int InstanceChecker::do_table_stream_check() {
+ struct VersionedOffset {
+ Versionstamp versionstamp;
+ TableStreamOffsetPB offset;
+ };
+
+ auto decode_components =
+ [](std::string_view key, size_t expected_size,
+ std::vector<std::tuple<std::variant<int64_t, std::string>, int,
int>>* components) {
+ if (key.empty()) {
+ return false;
+ }
+ key.remove_prefix(1);
+ return decode_key(&key, components) == 0 && components->size()
== expected_size;
+ };
+
+ std::unordered_map<int64_t, int> recycling_streams;
+ auto is_recycling = [&](int64_t stream_id) {
+ auto cached = recycling_streams.find(stream_id);
+ if (cached != recycling_streams.end()) {
+ return cached->second;
+ }
+ const int existence =
+ key_exist(txn_kv_.get(), recycle_index_key({instance_id_,
stream_id}));
+ const int result = existence < 0 ? -1 : existence == 0;
+ recycling_streams.emplace(stream_id, result);
+ return result;
+ };
+
+ int check_ret = 0;
+ std::unordered_map<std::string, TableStreamOffsetPB> latest_offsets;
+ auto validate_offset = [&](int64_t base_db_id, int64_t base_table_id,
int64_t stream_db_id,
+ int64_t stream_id, int64_t partition_id,
+ const TableStreamOffsetPB& offset) {
+ if (base_db_id <= 0 || base_table_id <= 0 || stream_db_id <= 0 ||
stream_id <= 0 ||
+ partition_id <= 0 || !offset.has_partition_id() ||
!offset.has_state() ||
+ !offset.has_offset_tso() || offset.partition_id() != partition_id
||
+ (offset.state() != TABLE_STREAM_OFFSET_INITIAL_SNAPSHOT_PENDING &&
+ offset.state() != TABLE_STREAM_OFFSET_CONSUMED)) {
+ LOG_WARNING("Table Stream Offset value does not match its key")
+ .tag("instance_id", instance_id_)
+ .tag("stream_id", stream_id)
+ .tag("key_partition_id", partition_id)
+ .tag("value_partition_id", offset.partition_id())
+ .tag("state", offset.state());
+ return 1;
+ }
+
+ const int recycling = is_recycling(stream_id);
+ if (recycling < 0) {
+ return -1;
+ }
+ if (recycling > 0) {
+ return 2;
+ }
+ return 0;
+ };
+
+ std::string begin = table_stream_offset_key({instance_id_, 0, 0, 0, 0, 0});
+ const std::string latest_end = table_stream_offset_key({instance_id_,
INT64_MAX, 0, 0, 0, 0});
+ int ret = scan_and_handle_kv(
+ begin, latest_end, [&](std::string_view key, std::string_view
value) {
+ std::vector<std::tuple<std::variant<int64_t, std::string>,
int, int>> components;
+ if (!decode_components(key, 8, &components)) {
+ LOG_WARNING("failed to decode Latest Stream Offset
key").tag("key", hex(key));
+ return -1;
+ }
+ TableStreamOffsetPB offset;
+ if (!offset.ParseFromArray(value.data(), value.size())) {
+ LOG_WARNING("failed to parse Latest Stream
Offset").tag("key", hex(key));
+ return -1;
+ }
+ int64_t base_db_id =
std::get<int64_t>(std::get<0>(components[3]));
+ int64_t base_table_id =
std::get<int64_t>(std::get<0>(components[4]));
+ int64_t stream_db_id =
std::get<int64_t>(std::get<0>(components[5]));
+ int64_t stream_id =
std::get<int64_t>(std::get<0>(components[6]));
+ int64_t partition_id =
std::get<int64_t>(std::get<0>(components[7]));
+ int validation = validate_offset(base_db_id, base_table_id,
stream_db_id, stream_id,
+ partition_id, offset);
+ if (validation == 2) {
+ return 0;
+ }
+ if (validation != 0) {
+ return validation;
+ }
+ latest_offsets.emplace(std::string(key), std::move(offset));
+ return 0;
+ });
+ if (ret < 0) {
+ return ret;
+ }
+ check_ret = std::max(check_ret, ret);
+
+ std::unordered_map<std::string, VersionedOffset> versioned_offsets;
+ begin = versioned::table_stream_offset_key({instance_id_, 0, 0, 0, 0, 0});
+ const std::string versioned_end =
+ versioned::table_stream_offset_key({instance_id_, INT64_MAX, 0, 0,
0, 0});
+ ret = scan_and_handle_kv(
Review Comment:
[P2] Compare the two projections at one snapshot
`scan_and_handle_kv` creates a fresh transaction for each call, so the
Latest and Versioned maps are not a coherent snapshot. A normal consumer can
advance both projections from 100 to 110 after the first scan: this code then
compares Latest=100 from RV1 with Versioned=110 from RV2 and marks the checker
job failed even though the atomic update is healthy. Use one stable read
version for both scans, or revalidate every apparent mismatch from one fresh
transaction before reporting corruption.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java:
##########
@@ -4003,6 +4008,7 @@ public void createTableStream(CreateStreamCommand
command) throws DdlException {
throw new DdlException("Unknown properties: " +
properties);
}
newStream.setId((Env.getCurrentEnv().getNextId()));
+ beforeCreateTableStream(db, newStream, baseTable);
Review Comment:
[P1] Make Stream publication recoverable before committing MetaService state
This hook fully commits the new Stream and removes its PREPARED recycle
marker before `createTableWithLock` arbitrates the name or journals the FE
entry. Two concurrent same-name creates can therefore commit distinct IDs; only
one is registered (and `IF NOT EXISTS` even returns success for the loser). A
master crash in the same gap leaves the same permanent MetaService-only Stream,
with no catalog owner or recycle key to clean it up. Add a durable pending/name
reservation whose replay can finish or drop the ID before final activation; an
in-process rollback alone will not close the crash window.
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/stream/TableStreamManager.java:
##########
@@ -377,6 +416,114 @@ public void
fillStreamConsumptionValuesMetadataResult(List<TRow> dataBatch) {
}
}
+ private void fillCloudStreamConsumptionValuesMetadataResult(List<TRow>
dataBatch) throws UserException {
+ Map<Cloud.TableStreamIdentityPB, CloudStreamConsumptionSnapshot>
snapshots = new LinkedHashMap<>();
+ for (Map.Entry<Long, Set<Long>> entry : copyDbStreamMap().entrySet()) {
+ Optional<Database> db =
Env.getCurrentInternalCatalog().getDb(entry.getKey());
+ if (!db.isPresent()) {
+ continue;
+ }
+ for (Long tableId : entry.getValue()) {
+ Optional<Table> table = db.get().getTable(tableId);
+ if (!table.isPresent()) {
+ continue;
+ }
+ Preconditions.checkArgument(table.get() instanceof
OlapTableStream);
+ OlapTableStream stream = (OlapTableStream) table.get();
+ if (!stream.readLockIfExist()) {
+ continue;
+ }
+ try {
+ OlapTable baseTable = stream.getBaseTableNullable();
+ if (baseTable == null || !baseTable.readLockIfExist()) {
+ continue;
+ }
+ try {
+ Map<Long, String> partitionNames = new
LinkedHashMap<>();
+ baseTable.getPartitions().forEach(partition ->
+ partitionNames.put(partition.getId(),
partition.getName()));
+ Cloud.TableStreamIdentityPB identity =
Cloud.TableStreamIdentityPB.newBuilder()
+
.setBaseDbId(stream.getBaseTableInfo().getDbId())
+
.setBaseTableId(stream.getBaseTableInfo().getTableId())
+ .setStreamDbId(entry.getKey())
+ .setStreamId(stream.getId())
+ .build();
+ CloudStreamConsumptionSnapshot previous =
snapshots.put(identity,
+ new
CloudStreamConsumptionSnapshot(db.get().getFullName(), stream.getName(),
+ stream.getId(), partitionNames));
+ Preconditions.checkState(previous == null,
+ "Duplicate Cloud Table Stream identity %s",
identity);
+ } finally {
+ baseTable.readUnlock();
+ }
+ } finally {
+ stream.readUnlock();
+ }
+ }
+ }
+ if (snapshots.isEmpty()) {
+ return;
+ }
+
+ Map<Cloud.TableStreamIdentityPB, Set<Long>> requestedPartitions = new
LinkedHashMap<>();
+ snapshots.forEach((identity, snapshot) ->
+ requestedPartitions.put(identity,
snapshot.partitionNames.keySet()));
Review Comment:
[P2] Omit Streams with no current partitions from this batch
A valid Stream can have an empty `partitionNames` map after all formal
partitions are dropped, but this code still sends its identity with an empty
partition set. MetaService deliberately rejects such a binding as
`INVALID_ARGUMENT`; because every Stream is fetched in one RPC, that one empty
Stream makes the whole consumption system-table query fail and hides rows for
unrelated Streams. Skip empty sets here and return normally when all sets are
empty; add a mixed empty/nonempty Stream test.
##########
cloud/test/meta_service_table_stream_test.cpp:
##########
@@ -0,0 +1,1006 @@
+// 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 <brpc/controller.h>
+#include <gen_cpp/cloud.pb.h>
+#include <gtest/gtest.h>
+
+#include <cstdint>
+#include <initializer_list>
+#include <limits>
+#include <memory>
+#include <optional>
+#include <string>
+#include <string_view>
+#include <vector>
+
+#include "common/config.h"
+#include "common/defer.h"
+#include "common/lexical_util.h"
+#include "meta-service/meta_service.h"
+#include "meta-store/keys.h"
+#include "meta-store/txn_kv.h"
+#include "meta-store/txn_kv_error.h"
+#include "meta-store/versioned_value.h"
+#include "recycler/recycler.h"
+#include "resource-manager/resource_manager.h"
+#include "snapshot/snapshot_manager.h"
+
+namespace doris::cloud {
+
+extern std::unique_ptr<MetaServiceProxy> get_meta_service(bool
mock_resource_mgr);
+extern TxnErrorCode read_operation_log(Transaction* txn, std::string_view
log_key,
+ Versionstamp* log_version,
OperationLogPB* operation_log);
+
+class MetaServiceTableStreamTest : public ::testing::Test {
+protected:
+ void SetUp() override {
+ service_ = get_meta_service(false);
+ instance_id_ = "table_stream_read_state_instance";
+ cloud_unique_id_ = "1:" + instance_id_ + ":test";
+ identity_.set_base_db_id(1001);
+ identity_.set_base_table_id(1002);
+ identity_.set_stream_db_id(1003);
+ identity_.set_stream_id(1004);
+ }
+
+ void set_multi_version_status(MultiVersionStatus mode) {
+ multi_version_status_ = mode;
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
+ InstanceInfoPB instance;
+ instance.set_instance_id(instance_id_);
+ instance.set_multi_version_status(mode);
+ txn->put(instance_key({instance_id_}), instance.SerializeAsString());
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+ auto [code, message] =
service_->resource_mgr()->refresh_instance(instance_id_);
+ ASSERT_EQ(code, MetaServiceCode::OK) << message;
+ }
+
+ void put_partition_mapping(Transaction* txn, int64_t partition_id) {
+ PartitionIndexPB partition;
+ partition.set_db_id(identity_.base_db_id());
+ partition.set_table_id(identity_.base_table_id());
+ txn->put(versioned::partition_index_key({instance_id_, partition_id}),
+ partition.SerializeAsString());
+ }
+
+ void put_latest_partition_state(Transaction* txn, int64_t partition_id,
int64_t visible_version,
+ int64_t visible_tso,
std::optional<int64_t> offset_tso) {
+ VersionPB version;
+ version.set_version(visible_version);
+ version.set_visible_tso(visible_tso);
+ txn->put(partition_version_key({instance_id_, identity_.base_db_id(),
+ identity_.base_table_id(),
partition_id}),
+ version.SerializeAsString());
+ if (offset_tso.has_value()) {
+ TableStreamOffsetPB offset;
+ offset.set_partition_id(partition_id);
+ offset.set_state(TABLE_STREAM_OFFSET_CONSUMED);
+ offset.set_offset_tso(*offset_tso);
+ offset.set_last_consumption_time_ms(1234);
+ txn->put(table_stream_offset_key({instance_id_,
identity_.base_db_id(),
+ identity_.base_table_id(),
identity_.stream_db_id(),
+ identity_.stream_id(),
partition_id}),
+ offset.SerializeAsString());
+ }
+ }
+
+ void put_versioned_partition_state(Transaction* txn, int64_t partition_id,
+ int64_t visible_version, int64_t
visible_tso,
+ std::optional<int64_t> offset_tso) {
+ put_partition_mapping(txn, partition_id);
+ versioned_put(txn, versioned::meta_partition_key({instance_id_,
partition_id}),
+ Versionstamp(41, 0), "");
+
+ VersionPB version;
+ version.set_version(visible_version);
+ version.set_visible_tso(visible_tso);
+ versioned_put(txn, versioned::partition_version_key({instance_id_,
partition_id}),
+ Versionstamp(42, 0), version.SerializeAsString());
+ if (offset_tso.has_value()) {
+ TableStreamOffsetPB offset;
+ offset.set_partition_id(partition_id);
+ offset.set_state(TABLE_STREAM_OFFSET_INITIAL_SNAPSHOT_PENDING);
+ offset.set_offset_tso(*offset_tso);
+ versioned_put(txn,
+ versioned::table_stream_offset_key(
+ {instance_id_, identity_.base_db_id(),
identity_.base_table_id(),
+ identity_.stream_db_id(),
identity_.stream_id(), partition_id}),
+ Versionstamp(43, 0), offset.SerializeAsString());
+ }
+ }
+
+ GetTableStreamOffsetResponse get_read_state(const std::vector<int64_t>&
partitions) {
+ GetTableStreamOffsetRequest request;
+ request.set_cloud_unique_id(cloud_unique_id_);
+ auto* binding = request.add_bindings();
+ binding->mutable_identity()->CopyFrom(identity_);
+ for (int64_t partition_id : partitions) {
+ binding->add_partition_ids(partition_id);
+ }
+ GetTableStreamOffsetResponse response;
+ brpc::Controller controller;
+ service_->get_table_stream_offset(&controller, &request, &response,
nullptr);
+ return response;
+ }
+
+ GetTableStreamOffsetResponse get_read_state(std::initializer_list<int64_t>
partitions) {
+ return get_read_state(std::vector<int64_t>(partitions));
+ }
+
+ int64_t begin_target_transaction(const std::string& label) {
+ BeginTxnRequest request;
+ request.set_cloud_unique_id(cloud_unique_id_);
+ TxnInfoPB* txn_info = request.mutable_txn_info();
+ txn_info->set_db_id(3001);
+ txn_info->set_label(label);
+ txn_info->add_table_ids(3002);
+ txn_info->set_timeout_ms(36000);
+ BeginTxnResponse response;
+ brpc::Controller controller;
+ service_->begin_txn(&controller, &request, &response, nullptr);
+ EXPECT_EQ(response.status().code(), MetaServiceCode::OK) <<
response.status().msg();
+ return response.txn_id();
+ }
+
+ CommitTxnRequest make_consume_request(int64_t txn_id, int64_t partition_id,
+ TableStreamOffsetStatePB
expected_state,
+ int64_t expected_tso, int64_t
next_tso) {
+ CommitTxnRequest request;
+ request.set_cloud_unique_id(cloud_unique_id_);
+ request.set_db_id(3001);
+ request.set_txn_id(txn_id);
+ TableStreamUpdatePB* stream_update =
request.add_table_stream_updates();
+ stream_update->mutable_identity()->CopyFrom(identity_);
+ TableStreamPartitionUpdatePB* partition_update =
stream_update->add_partition_updates();
+ partition_update->set_partition_id(partition_id);
+ partition_update->set_expected_state(expected_state);
+ if (expected_state != TABLE_STREAM_OFFSET_UNKNOWN) {
+ partition_update->set_expected_offset_tso(expected_tso);
+ }
+ partition_update->set_next_offset_tso(next_tso);
+ return request;
+ }
+
+ CommitTxnResponse commit_transaction(const CommitTxnRequest& request) {
+ CommitTxnResponse response;
+ brpc::Controller controller;
+ service_->commit_txn(&controller, &request, &response, nullptr);
+ return response;
+ }
+
+ CommitTxnResponse consume_partition(int64_t txn_id, int64_t partition_id,
+ TableStreamOffsetStatePB
expected_state,
+ int64_t expected_tso, int64_t
next_tso) {
+ return commit_transaction(
+ make_consume_request(txn_id, partition_id, expected_state,
expected_tso, next_tso));
+ }
+
+ TableStreamOffsetPB get_latest_offset(const TableStreamIdentityPB&
identity,
+ int64_t partition_id) {
+ std::unique_ptr<Transaction> txn;
+ EXPECT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
+ std::string value;
+ EXPECT_EQ(txn->get(table_stream_offset_key(
+ {instance_id_, identity.base_db_id(),
identity.base_table_id(),
+ identity.stream_db_id(),
identity.stream_id(), partition_id}),
+ &value),
+ TxnErrorCode::TXN_OK);
+ TableStreamOffsetPB offset;
+ EXPECT_TRUE(offset.ParseFromString(value));
+ return offset;
+ }
+
+ TableStreamOffsetPB get_latest_offset(int64_t partition_id) {
+ return get_latest_offset(identity_, partition_id);
+ }
+
+ void set_clone_source(const std::string& source_instance_id, Versionstamp
snapshot_version) {
+ multi_version_status_ = MULTI_VERSION_READ_WRITE;
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
+ InstanceInfoPB instance;
+ instance.set_instance_id(instance_id_);
+ instance.set_multi_version_status(MULTI_VERSION_READ_WRITE);
+ instance.set_source_instance_id(source_instance_id);
+
instance.set_source_snapshot_id(SnapshotManager::serialize_snapshot_id(snapshot_version));
+ txn->put(instance_key({instance_id_}), instance.SerializeAsString());
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+ auto [code, message] =
service_->resource_mgr()->refresh_instance(instance_id_);
+ ASSERT_EQ(code, MetaServiceCode::OK) << message;
+ }
+
+ Versionstamp put_auto_versioned_offset(const std::string&
target_instance_id,
+ const TableStreamIdentityPB&
identity,
+ int64_t partition_id, int64_t
offset_tso,
+ bool write_latest = false) {
+ std::unique_ptr<Transaction> txn;
+ EXPECT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
+ TableStreamOffsetPB offset;
+ offset.set_partition_id(partition_id);
+ offset.set_state(TABLE_STREAM_OFFSET_CONSUMED);
+ offset.set_offset_tso(offset_tso);
+ const TableStreamOffsetKeyInfo key_info {target_instance_id,
identity.base_db_id(),
+ identity.base_table_id(),
identity.stream_db_id(),
+ identity.stream_id(),
partition_id};
+ const std::string value = offset.SerializeAsString();
+ if (write_latest) {
+ txn->put(table_stream_offset_key(key_info), value);
+ }
+ txn->enable_get_versionstamp();
+ versioned_put(txn.get(), versioned::table_stream_offset_key(key_info),
value);
+ EXPECT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+ Versionstamp version;
+ EXPECT_EQ(txn->get_versionstamp(&version), TxnErrorCode::TXN_OK);
+ return version;
+ }
+
+ std::unique_ptr<MetaServiceProxy> service_;
+ std::string instance_id_;
+ std::string cloud_unique_id_;
+ TableStreamIdentityPB identity_;
+ MultiVersionStatus multi_version_status_ = MULTI_VERSION_DISABLED;
+};
+
+TEST_F(MetaServiceTableStreamTest, ReadLatestAndUnknownOffsets) {
+ set_multi_version_status(MULTI_VERSION_DISABLED);
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
+ put_latest_partition_state(txn.get(), 2001, 8, 130, 100);
+ put_latest_partition_state(txn.get(), 2002, 9, 140, std::nullopt);
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+
+ GetTableStreamOffsetResponse response = get_read_state({2001, 2002});
+ ASSERT_EQ(response.status().code(), MetaServiceCode::OK) <<
response.status().msg();
+ ASSERT_EQ(response.bindings_size(), 1);
+ ASSERT_EQ(response.bindings(0).partition_states_size(), 2);
+
+ const auto& consumed = response.bindings(0).partition_states(0);
+ EXPECT_EQ(consumed.partition_id(), 2001);
+ EXPECT_EQ(consumed.offset_state(), TABLE_STREAM_OFFSET_CONSUMED);
+ EXPECT_EQ(consumed.offset_tso(), 100);
+ EXPECT_EQ(consumed.end_tso(), 130);
+ EXPECT_EQ(consumed.visible_version(), 8);
+ EXPECT_EQ(consumed.last_consumption_time_ms(), 1234);
+
+ const auto& unknown = response.bindings(0).partition_states(1);
+ EXPECT_EQ(unknown.partition_id(), 2002);
+ EXPECT_EQ(unknown.offset_state(), TABLE_STREAM_OFFSET_UNKNOWN);
+ EXPECT_FALSE(unknown.has_offset_tso());
+ EXPECT_EQ(unknown.end_tso(), 140);
+ EXPECT_EQ(unknown.visible_version(), 9);
+}
+
+TEST_F(MetaServiceTableStreamTest, ReadVersionedState) {
+ set_multi_version_status(MULTI_VERSION_READ_WRITE);
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
+ put_versioned_partition_state(txn.get(), 2001, 18, 230, 200);
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+
+ GetTableStreamOffsetResponse response = get_read_state({2001});
+ ASSERT_EQ(response.status().code(), MetaServiceCode::OK) <<
response.status().msg();
+ ASSERT_EQ(response.bindings_size(), 1);
+ ASSERT_EQ(response.bindings(0).partition_states_size(), 1);
+ const auto& state = response.bindings(0).partition_states(0);
+ EXPECT_EQ(state.offset_state(),
TABLE_STREAM_OFFSET_INITIAL_SNAPSHOT_PENDING);
+ EXPECT_EQ(state.offset_tso(), 200);
+ EXPECT_EQ(state.end_tso(), 230);
+ EXPECT_EQ(state.visible_version(), 18);
+}
+
+TEST_F(MetaServiceTableStreamTest, ReadVersionedStateFromCloneChainInBatch) {
+ const std::string source_instance_id = "table_stream_batch_read_source";
+ set_clone_source(source_instance_id, Versionstamp(100, 0));
+ const std::vector<int64_t> partition_ids = {2001, 2002, 2003};
+
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
+ for (size_t i = 0; i < partition_ids.size(); ++i) {
+ int64_t partition_id = partition_ids[i];
+ PartitionIndexPB partition;
+ partition.set_db_id(identity_.base_db_id());
+ partition.set_table_id(identity_.base_table_id());
+ txn->put(versioned::partition_index_key({source_instance_id,
partition_id}),
+ partition.SerializeAsString());
+ versioned_put(txn.get(),
versioned::meta_partition_key({source_instance_id, partition_id}),
+ Versionstamp(51 + i, 0), "");
+
+ VersionPB version;
+ version.set_version(10 + i);
+ version.set_visible_tso(1000 + i);
+ versioned_put(txn.get(),
+ versioned::partition_version_key({source_instance_id,
partition_id}),
+ Versionstamp(60 + i, 0), version.SerializeAsString());
+ }
+
+ auto put_offset = [&](const std::string& target_instance_id, int64_t
partition_id,
+ Versionstamp versionstamp, int64_t offset_tso) {
+ TableStreamOffsetPB offset;
+ offset.set_partition_id(partition_id);
+ offset.set_state(TABLE_STREAM_OFFSET_CONSUMED);
+ offset.set_offset_tso(offset_tso);
+ versioned_put(
+ txn.get(),
+ versioned::table_stream_offset_key(
+ {target_instance_id, identity_.base_db_id(),
identity_.base_table_id(),
+ identity_.stream_db_id(), identity_.stream_id(),
partition_id}),
+ versionstamp, offset.SerializeAsString());
+ };
+ put_offset(source_instance_id, 2001, Versionstamp(70, 0), 701);
+ put_offset(source_instance_id, 2002, Versionstamp(71, 0), 702);
+ put_offset(instance_id_, 2002, Versionstamp(200, 0), 802);
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+
+ GetTableStreamOffsetResponse response = get_read_state(partition_ids);
+ ASSERT_EQ(response.status().code(), MetaServiceCode::OK) <<
response.status().msg();
+ ASSERT_EQ(response.bindings(0).partition_states_size(),
partition_ids.size());
+ EXPECT_EQ(response.bindings(0).partition_states(0).offset_tso(), 701);
+ EXPECT_EQ(response.bindings(0).partition_states(1).offset_tso(), 802);
+ EXPECT_EQ(response.bindings(0).partition_states(2).offset_state(),
TABLE_STREAM_OFFSET_UNKNOWN);
+}
+
+TEST_F(MetaServiceTableStreamTest, ReadLargePartitionBatch) {
+ set_multi_version_status(MULTI_VERSION_DISABLED);
+ constexpr int kPartitionCount = 1201;
+ std::vector<int64_t> partition_ids;
+ partition_ids.reserve(kPartitionCount);
+
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
+ for (int i = 0; i < kPartitionCount; ++i) {
+ int64_t partition_id = 10000 + i;
+ partition_ids.push_back(partition_id);
+ put_latest_partition_state(txn.get(), partition_id, i + 1, 100000 + i,
90000 + i);
+ }
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+
+ GetTableStreamOffsetResponse response = get_read_state(partition_ids);
+ ASSERT_EQ(response.status().code(), MetaServiceCode::OK) <<
response.status().msg();
+ ASSERT_EQ(response.bindings(0).partition_states_size(), kPartitionCount);
+ EXPECT_EQ(response.bindings(0).partition_states(0).partition_id(),
partition_ids.front());
+ EXPECT_EQ(response.bindings(0).partition_states(0).offset_tso(), 90000);
+ EXPECT_EQ(response.bindings(0).partition_states(kPartitionCount -
1).partition_id(),
+ partition_ids.back());
+ EXPECT_EQ(response.bindings(0).partition_states(kPartitionCount -
1).offset_tso(),
+ 90000 + kPartitionCount - 1);
+}
+
+TEST_F(MetaServiceTableStreamTest, ReadMultipleBindingsInBatch) {
+ set_multi_version_status(MULTI_VERSION_READ_WRITE);
+ constexpr int kBindingCount = 101;
+ constexpr int64_t kPartitionId = 2001;
+
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
+ PartitionIndexPB partition;
+ partition.set_db_id(identity_.base_db_id());
+ partition.set_table_id(identity_.base_table_id());
+ txn->put(versioned::partition_index_key({instance_id_, kPartitionId}),
+ partition.SerializeAsString());
+ versioned_put(txn.get(), versioned::meta_partition_key({instance_id_,
kPartitionId}),
+ Versionstamp(20, 0), "");
+ VersionPB version;
+ version.set_version(8);
+ version.set_visible_tso(130);
+ versioned_put(txn.get(), versioned::partition_version_key({instance_id_,
kPartitionId}),
+ Versionstamp(21, 0), version.SerializeAsString());
+
+ GetTableStreamOffsetRequest request;
+ request.set_cloud_unique_id(cloud_unique_id_);
+ for (int i = 0; i < kBindingCount; ++i) {
+ int64_t stream_id = identity_.stream_id() + i;
+ TableStreamPartitionSetPB* binding = request.add_bindings();
+ binding->mutable_identity()->CopyFrom(identity_);
+ binding->mutable_identity()->set_stream_id(stream_id);
+ binding->add_partition_ids(kPartitionId);
+ }
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+
+ GetTableStreamOffsetResponse response;
+ brpc::Controller controller;
+ service_->get_table_stream_offset(&controller, &request, &response,
nullptr);
+ ASSERT_EQ(response.status().code(), MetaServiceCode::OK) <<
response.status().msg();
+ ASSERT_EQ(response.bindings_size(), kBindingCount);
+ for (const TableStreamReadBindingResultPB& binding : response.bindings()) {
+ ASSERT_EQ(binding.partition_states_size(), 1);
+ EXPECT_EQ(binding.partition_states(0).offset_state(),
TABLE_STREAM_OFFSET_UNKNOWN);
+ }
+}
+
+TEST_F(MetaServiceTableStreamTest, ReadWriteOnlyState) {
+ set_multi_version_status(MULTI_VERSION_WRITE_ONLY);
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
+ put_latest_partition_state(txn.get(), 2001, 28, 330, 300);
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+
+ GetTableStreamOffsetResponse response = get_read_state({2001});
+ ASSERT_EQ(response.status().code(), MetaServiceCode::OK) <<
response.status().msg();
+ ASSERT_EQ(response.bindings_size(), 1);
+ ASSERT_EQ(response.bindings(0).partition_states_size(), 1);
+ const auto& state = response.bindings(0).partition_states(0);
+ EXPECT_EQ(state.offset_state(), TABLE_STREAM_OFFSET_CONSUMED);
+ EXPECT_EQ(state.offset_tso(), 300);
+ EXPECT_EQ(state.end_tso(), 330);
+ EXPECT_EQ(state.visible_version(), 28);
+}
+
+TEST_F(MetaServiceTableStreamTest, RejectDuplicateBindingsAndPartitions) {
+ GetTableStreamOffsetRequest request;
+ request.set_cloud_unique_id(cloud_unique_id_);
+ auto* binding = request.add_bindings();
+ binding->mutable_identity()->CopyFrom(identity_);
+ binding->add_partition_ids(2001);
+ binding->add_partition_ids(2001);
+ GetTableStreamOffsetResponse response;
+ brpc::Controller controller;
+ service_->get_table_stream_offset(&controller, &request, &response,
nullptr);
+ EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
+
+ request.mutable_bindings()->Clear();
+ for (int i = 0; i < 2; ++i) {
+ binding = request.add_bindings();
+ binding->mutable_identity()->CopyFrom(identity_);
+ binding->add_partition_ids(2001 + i);
+ }
+ response.Clear();
+ brpc::Controller second_controller;
+ service_->get_table_stream_offset(&second_controller, &request, &response,
nullptr);
+ EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
+}
+
+TEST_F(MetaServiceTableStreamTest, RejectEnabledModeAndRecyclingStream) {
+ set_multi_version_status(MULTI_VERSION_DISABLED);
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
+ InstanceInfoPB instance;
+ instance.set_instance_id(instance_id_);
+ instance.set_multi_version_status(MULTI_VERSION_ENABLED);
+ txn->put(instance_key({instance_id_}), instance.SerializeAsString());
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+
+ GetTableStreamOffsetResponse response = get_read_state({2001});
+ EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
+
+ set_multi_version_status(MULTI_VERSION_DISABLED);
+ ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
+ put_latest_partition_state(txn.get(), 2001, 8, 130, 100);
+ RecycleIndexPB recycle_index;
+ recycle_index.set_table_id(identity_.base_table_id());
+ recycle_index.set_state(RecycleIndexPB::RECYCLING);
+ recycle_index.set_object_type(TABLE_STREAM);
+ txn->put(recycle_index_key({instance_id_, identity_.stream_id()}),
+ recycle_index.SerializeAsString());
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+
+ response = get_read_state({2001});
+ EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
+}
+
+TEST_F(MetaServiceTableStreamTest, CommitUpdatesLatestOffsetAndIsIdempotent) {
Review Comment:
[P2] Exercise target-write/offset atomicity in this test
All consumption tests begin an otherwise empty target transaction, so they
only prove atomicity among offset keys. The feature's critical contract is that
a stale CAS publishes neither the target rowset nor the offset, while a valid
CAS publishes both in the same commit. Please prepare a real target tmp rowset
here, assert both sides remain invisible after a stale expected offset, then
commit a fresh transaction and assert the rowset and offset become visible
together (including retry/idempotency).
##########
fe/fe-core/src/main/java/org/apache/doris/cloud/datasource/CloudInternalCatalog.java:
##########
@@ -949,6 +1053,9 @@ public void dropCloudPartition(long dbId, long tableId,
List<Long> partitionIds,
partitionRequestBuilder.setTableId(tableId);
partitionRequestBuilder.addAllPartitionIds(partitionIds);
partitionRequestBuilder.addAllIndexIds(indexIds);
+ partitionRequestBuilder.addAllTableStreams(
Review Comment:
[P2] Coordinate partition drops with pending Stream creation
The Stream's MetaService offsets are committed while CREATE holds only the
base-table read lock, and the FE entry is published later under the Stream
database lock. A concurrent `DROP PARTITION ... FORCE` can already hold the
base database lock, run as soon as that read lock is released, and enumerate
Stream identities here before the new entry is visible. Its operation log then
omits the committed Stream offset; CREATE publishes afterward with a stale
Latest/Versioned offset for the deleted partition, and no later Cloud cleanup
removes it. Make pending Stream identities visible to base-table DDL, or
revalidate and delete offsets for partitions removed before activation.
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/CatalogRecycleBin.java:
##########
@@ -463,15 +474,22 @@ private void eraseTable(long currentTimeMs, int keepNum) {
for (Long tableId : expiredIds) {
writeLock();
try {
- RecycleTableInfo tableInfo = idToTable.remove(tableId);
+ RecycleTableInfo tableInfo = idToTable.get(tableId);
if (tableInfo == null) {
continue;
}
Table table = tableInfo.getTable();
+ try {
+
Env.getCurrentInternalCatalog().beforeEraseTable(tableInfo.dbId, table, false);
Review Comment:
[P1] Move the remote Stream drop outside the global recycle-bin lock
For Cloud Streams this hook synchronously calls a retrying `dropIndex`, but
this call and the other three new call sites all run under the recycle bin's
single write lock. A slow/unavailable MetaService therefore blocks unrelated
drops, recoveries, recycle inserts, and even read-side inspection for the
nested retry/deadline window. Reserve or durably stage the exact entry while
locked, perform the idempotent RPC after unlocking, then reacquire and verify
the reservation before logging/removing it so recovery cannot race the
completed remote drop.
##########
fe/fe-core/src/main/java/org/apache/doris/cloud/datasource/CloudInternalCatalog.java:
##########
@@ -98,6 +103,140 @@ public CloudInternalCatalog() {
super();
}
+ @Override
+ protected void setTableStreamProperties(BaseTableStream stream,
Map<String, String> properties)
+ throws AnalysisException {
+ if (stream instanceof OlapTableStream) {
+ ((OlapTableStream)
stream).setPropertiesWithoutOffsetInitialization(properties);
+ return;
+ }
+ super.setTableStreamProperties(stream, properties);
+ }
+
+ @Override
+ protected void beforeCreateTableStream(Database streamDb, BaseTableStream
stream, TableIf baseTable)
+ throws DdlException {
+ if (!(stream instanceof OlapTableStream) || !(baseTable instanceof
OlapTable)) {
+ throw new DdlException("Cloud Table Stream requires an OLAP base
table");
+ }
+ OlapTableStream olapStream = (OlapTableStream) stream;
+ OlapTable olapBaseTable = (OlapTable) baseTable;
+ List<Cloud.TableStreamOffsetPB> initialOffsets =
captureTableStreamInitialOffsets(
+ olapStream, olapBaseTable);
+ Set<Long> basePartitionIds = new
HashSet<>(olapBaseTable.getPartitionIds());
+ Set<Long> offsetPartitionIds = initialOffsets.stream()
+ .map(Cloud.TableStreamOffsetPB::getPartitionId)
+ .collect(Collectors.toSet());
+ if (initialOffsets.size() != offsetPartitionIds.size()
+ || !basePartitionIds.equals(offsetPartitionIds)) {
+ throw new DdlException("Cloud Table Stream initial offsets do not
match base table partitions");
+ }
+
+ long baseDbId = olapStream.getBaseTableInfo().getDbId();
+ prepareTableStream(baseDbId, olapBaseTable.getId(), streamDb.getId(),
olapStream.getId());
Review Comment:
[P2] Release the base-table lock before retrying MetaService RPCs
`createTableStream` invokes this hook while holding `baseTable.readLock()`,
and this call starts a sequence of prepare, partition-batch, and final-commit
RPCs. Both this helper and `MetaServiceProxy` can retry and sleep, so an
unhealthy MetaService holds the table lock across multiple deadline windows and
blocks unrelated ALTER/DROP operations. Capture and validate an immutable
base-table snapshot under the lock, then perform the remote work after
unlocking, with the pending-create/revalidation state needed to detect
intervening metadata changes.
##########
cloud/src/meta-service/meta_service_partition.cpp:
##########
@@ -83,6 +88,121 @@ 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;
+}
+
+template <typename Request>
+static bool table_stream_recycle_index_matches(const RecycleIndexPB&
recycle_index,
+ const Request& 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;
+ }
+ // One returned key is sufficient for existence. `more()` only means that
additional
+ // partition offsets remain in the range.
+ return it->has_next() ? TxnErrorCode::TXN_OK :
TxnErrorCode::TXN_KEY_NOT_FOUND;
+}
+
+static TxnErrorCode get_table_stream_min_local_offset_version(
+ 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, Versionstamp* min_version) {
+ // Stream recycling removes only the current instance's offset prefix.
Keep this range read in
+ // the DROP transaction so an offset write committed before this
transaction cannot escape the
+ // min_timestamp calculation. A later recycle index rejects writes after
it is materialized.
+ const std::string begin = versioned::table_stream_offset_key_prefix(
+ instance_id, base_db_id, base_table_id, stream_db_id, stream_id);
+ FullRangeGetOptions options;
+ options.snapshot = false;
+ options.prefetch = true;
+ auto iter = txn->full_range_get(begin, lexical_end(begin), options);
Review Comment:
[P2] Avoid scanning the full offset history in the DROP transaction
This binds `full_range_get` to the same FoundationDB transaction that must
later install the drop log, then walks every retained Versioned offset. Every
consumption adds more keys and active snapshots intentionally retain old
versions, so a large Stream history can exceed the transaction timeout before
this loop completes. Retrying repeats the same unbounded scan, leaving the
Stream undroppable. Maintain a conservative per-Stream earliest-local-version
summary with offset writes (a stale-low value is safe), conflict-read that
bounded key during DROP, and clear it during final recycling.
##########
cloud/src/meta-service/meta_service_txn.cpp:
##########
@@ -48,6 +54,280 @@ namespace doris::cloud {
static constexpr std::string_view kMetaSyncPointDummyKey =
"__meta_service_sync_point_dummy_key__";
+static bool validate_table_stream_updates(const CommitTxnRequest* request,
MetaServiceCode& code,
+ std::string& msg) {
+ if (request->has_is_2pc() && request->is_2pc()) {
+ code = MetaServiceCode::INVALID_ARGUMENT;
+ msg = "table stream consumption does not support 2PC";
+ return false;
+ }
+ if ((request->has_is_txn_load() && request->is_txn_load()) ||
+ !request->sub_txn_infos().empty()) {
+ code = MetaServiceCode::INVALID_ARGUMENT;
+ msg = "table stream consumption does not support transaction load or
sub transactions";
+ return false;
+ }
+
+ std::set<std::pair<int64_t, int64_t>> stream_partitions;
+ for (const auto& stream_update : request->table_stream_updates()) {
+ if (!stream_update.has_identity() ||
+ !is_valid_table_stream_identity(stream_update.identity()) ||
+ stream_update.partition_updates().empty()) {
+ code = MetaServiceCode::INVALID_ARGUMENT;
+ msg = "invalid table stream update";
+ return false;
+ }
+ for (const auto& partition_update : stream_update.partition_updates())
{
+ if (!partition_update.has_partition_id() ||
partition_update.partition_id() <= 0 ||
+ !partition_update.has_expected_state() ||
!partition_update.has_next_offset_tso()) {
+ code = MetaServiceCode::INVALID_ARGUMENT;
+ msg = "invalid table stream partition update";
+ return false;
+ }
+ if (partition_update.expected_state() !=
+ TableStreamOffsetStatePB::TABLE_STREAM_OFFSET_UNKNOWN
&&
+ partition_update.expected_state() !=
+
TableStreamOffsetStatePB::TABLE_STREAM_OFFSET_INITIAL_SNAPSHOT_PENDING &&
+ partition_update.expected_state() !=
+
TableStreamOffsetStatePB::TABLE_STREAM_OFFSET_CONSUMED) {
+ code = MetaServiceCode::INVALID_ARGUMENT;
+ msg = "invalid expected table stream offset state";
+ return false;
+ }
+ bool expects_existing_offset = partition_update.expected_state() !=
+
TableStreamOffsetStatePB::TABLE_STREAM_OFFSET_UNKNOWN;
+ if (partition_update.has_expected_offset_tso() !=
expects_existing_offset) {
+ code = MetaServiceCode::INVALID_ARGUMENT;
+ msg = "expected offset TSO does not match table stream offset
state";
+ return false;
+ }
+ auto [_, inserted] =
stream_partitions.emplace(stream_update.identity().stream_id(),
+
partition_update.partition_id());
+ if (!inserted) {
+ code = MetaServiceCode::INVALID_ARGUMENT;
+ msg = "duplicate table stream partition update";
+ return false;
+ }
+ }
+ }
+ return true;
+}
+
+static void append_table_stream_commit_size_error(TxnErrorCode err,
std::string& msg) {
+ if (err == TxnErrorCode::TXN_BYTES_TOO_LARGE) {
+ msg += ", table stream offset updates cannot use lazy commit. "
+ "Please consume fewer partitions in one statement.";
+ }
+}
+
+class TableStreamUpdateTxnContext {
+public:
+ TableStreamUpdateTxnContext(Transaction* txn, const std::string&
instance_id,
+ MultiVersionStatus multi_version_status,
+ CloneChainReader* clone_reader,
MetaServiceCode& code,
+ std::string& msg)
+ : txn_(txn),
+ instance_id_(instance_id),
+ metadata_reader_(txn, instance_id, multi_version_status,
clone_reader),
+ code_(code),
+ msg_(msg),
+ consumption_time_ms_(
+
duration_cast<milliseconds>(system_clock::now().time_since_epoch()).count()) {
+ }
+
+ bool versioned_write() const { return
metadata_reader_.writes_versioned_metadata(); }
+
+ bool check_stream(const TableStreamIdentityPB& identity) {
+ const std::vector<int64_t> stream_ids {identity.stream_id()};
+ std::unordered_set<int64_t> recycling_stream_ids;
+ if (!apply_read_result(metadata_reader_.read_recycling_streams(
+ stream_ids, TableStreamReadIntent::CONFLICT,
&recycling_stream_ids))) {
+ return false;
+ }
+ if (recycling_stream_ids.contains(identity.stream_id())) {
Review Comment:
[P1] Keep DROP fenced until older consume transactions expire
This is the only lifecycle guard read by a consume commit, but final Stream
recycling deletes the key after removing both offset prefixes. A transaction
planned before DROP for a newly added partition carries
`expected_state=UNKNOWN`; once this marker is gone, the live base partition
still validates and absent-offset CAS succeeds, so the old target commit
recreates Latest/Versioned offsets for a Stream that no longer exists in FE.
The default dropped-index retention is 3 hours while valid load transactions
may run for up to 3 days. Retain a tombstone that commits conflict-read for at
least the maximum transaction lifetime (or durably for non-reused Stream IDs),
and test DROP plus full recycle before a delayed UNKNOWN-state commit.
--
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]
