This is an automated email from the ASF dual-hosted git repository.
luwei16 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 5f3321b4cac [feat](TSO) Improve TSO status access (#65850)
5f3321b4cac is described below
commit 5f3321b4caca562d870002aa6cd59f11423a5e23
Author: Jingzhe Jia <[email protected]>
AuthorDate: Fri Aug 7 11:45:04 2026 +0800
[feat](TSO) Improve TSO status access (#65850)
### What problem does this PR solve?
Issue Number: close #65849
Related PR: #xxx
Problem Summary:
The /api/tso endpoint only served requests on the master FE, so requests
sent to a follower failed instead of reaching the authoritative TSO
service. TSO status also needed a queryable system-table interface
without adding a dedicated SHOW command.
This change forwards /api/tso requests to the master FE by default, adds
local=true to inspect the receiving FE without forwarding, and reads all
response fields from one TSO status snapshot. It also exposes the master
TSO state through information_schema.tso_status using the existing
schema-table RPC path.
---
be/src/information_schema/schema_scanner.cpp | 3 +
.../schema_tso_status_scanner.cpp | 128 ++++++++++++
.../information_schema/schema_tso_status_scanner.h | 54 +++++
.../schema_tso_status_scanner_test.cpp | 228 +++++++++++++++++++++
.../org/apache/doris/analysis/SchemaTableType.java | 3 +-
.../java/org/apache/doris/catalog/SchemaTable.java | 10 +
.../org/apache/doris/httpv2/rest/TSOAction.java | 25 ++-
.../doris/tablefunction/MetadataGenerator.java | 39 ++++
.../main/java/org/apache/doris/tso/TSOService.java | 42 ++++
.../org/apache/doris/catalog/SchemaTableTest.java | 9 +
.../apache/doris/httpv2/rest/TSOActionTest.java | 90 ++++++++
.../TsoStatusMetadataGeneratorTest.java | 132 ++++++++++++
.../java/org/apache/doris/tso/TSOServiceTest.java | 5 +-
gensrc/thrift/Descriptors.thrift | 1 +
gensrc/thrift/FrontendService.thrift | 1 +
regression-test/suites/tso_p0/test_tso_api.groovy | 33 +++
16 files changed, 790 insertions(+), 13 deletions(-)
diff --git a/be/src/information_schema/schema_scanner.cpp
b/be/src/information_schema/schema_scanner.cpp
index 635ba04469a..b3229b74827 100644
--- a/be/src/information_schema/schema_scanner.cpp
+++ b/be/src/information_schema/schema_scanner.cpp
@@ -82,6 +82,7 @@
#include "information_schema/schema_table_streams_scanner.h"
#include "information_schema/schema_tables_scanner.h"
#include "information_schema/schema_tablets_scanner.h"
+#include "information_schema/schema_tso_status_scanner.h"
#include "information_schema/schema_user_privileges_scanner.h"
#include "information_schema/schema_user_scanner.h"
#include "information_schema/schema_variables_scanner.h"
@@ -299,6 +300,8 @@ std::unique_ptr<SchemaScanner>
SchemaScanner::create(TSchemaTableType::type type
return SchemaCompactionTasksScanner::create_unique();
case TSchemaTableType::SCH_BACKEND_MS_RPC_TABLE_THROTTLERS:
return SchemaBackendMsRpcTableThrottlersScanner::create_unique();
+ case TSchemaTableType::SCH_TSO_STATUS:
+ return SchemaTsoStatusScanner::create_unique();
default:
return SchemaDummyScanner::create_unique();
break;
diff --git a/be/src/information_schema/schema_tso_status_scanner.cpp
b/be/src/information_schema/schema_tso_status_scanner.cpp
new file mode 100644
index 00000000000..ef9a0584fef
--- /dev/null
+++ b/be/src/information_schema/schema_tso_status_scanner.cpp
@@ -0,0 +1,128 @@
+// 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 "information_schema/schema_tso_status_scanner.h"
+
+#include <gen_cpp/FrontendService_types.h>
+
+#include "core/block/block.h"
+#include "core/data_type/data_type_factory.hpp"
+#include "runtime/cluster_info.h"
+#include "runtime/exec_env.h"
+#include "runtime/runtime_state.h"
+#include "util/client_cache.h"
+#include "util/thrift_rpc_helper.h"
+
+namespace doris {
+
+std::vector<SchemaScanner::ColumnDesc>
SchemaTsoStatusScanner::_s_tso_status_columns = {
+ {"WINDOW_END_PHYSICAL_TIME", TYPE_BIGINT, sizeof(int64_t), true},
+ {"CURRENT_TSO", TYPE_BIGINT, sizeof(int64_t), true},
+ {"CURRENT_TSO_PHYSICAL_TIME", TYPE_BIGINT, sizeof(int64_t), true},
+ {"CURRENT_TSO_LOGICAL_COUNTER", TYPE_BIGINT, sizeof(int64_t), true},
+};
+
+SchemaTsoStatusScanner::SchemaTsoStatusScanner()
+ : SchemaScanner(_s_tso_status_columns,
TSchemaTableType::SCH_TSO_STATUS) {}
+
+SchemaTsoStatusScanner::~SchemaTsoStatusScanner() = default;
+
+Status SchemaTsoStatusScanner::start(RuntimeState* state) {
+ _block_rows_limit = state->batch_size();
+ _rpc_timeout_ms = state->execution_timeout() * 1000;
+ return Status::OK();
+}
+
+Status SchemaTsoStatusScanner::_get_tso_status_block_from_fe() {
+ TNetworkAddress master_addr =
ExecEnv::GetInstance()->cluster_info()->master_fe_addr;
+
+ TSchemaTableRequestParams schema_table_request_params;
+ TFetchSchemaTableDataRequest request;
+ request.__set_schema_table_name(TSchemaTableName::TSO_STATUS);
+ request.__set_schema_table_params(schema_table_request_params);
+
+ TFetchSchemaTableDataResult result;
+ RETURN_IF_ERROR(ThriftRpcHelper::rpc<FrontendServiceClient>(
+ master_addr.hostname, master_addr.port,
+ [&request, &result](FrontendServiceConnection& client) {
+ client->fetchSchemaTableData(result, request);
+ },
+ _rpc_timeout_ms));
+
+ return _process_tso_status_result(result);
+}
+
+Status SchemaTsoStatusScanner::_process_tso_status_result(
+ const TFetchSchemaTableDataResult& result) {
+ Status status(Status::create(result.status));
+ if (!status.ok()) {
+ LOG(WARNING) << "fetch TSO status from FE failed, errmsg=" << status;
+ return status;
+ }
+
+ _tso_status_block = Block::create_unique();
+ for (int i = 0; i < _s_tso_status_columns.size(); ++i) {
+ auto data_type =
+
DataTypeFactory::instance().create_data_type(_s_tso_status_columns[i].type,
true);
+
_tso_status_block->insert(ColumnWithTypeAndName(data_type->create_column(),
data_type,
+
_s_tso_status_columns[i].name));
+ }
+
+ _tso_status_block->reserve(result.data_batch.size());
+ for (const TRow& row : result.data_batch) {
+ if (row.column_value.size() != _s_tso_status_columns.size()) {
+ return Status::InternalError<false>(
+ "TSO status schema does not match between FE and BE");
+ }
+ for (int i = 0; i < _s_tso_status_columns.size(); ++i) {
+ RETURN_IF_ERROR(insert_block_column(row.column_value[i], i,
_tso_status_block.get(),
+
_s_tso_status_columns[i].type));
+ }
+ }
+ _total_rows = static_cast<int>(_tso_status_block->rows());
+ return Status::OK();
+}
+
+Status SchemaTsoStatusScanner::get_next_block_internal(Block* block, bool*
eos) {
+ if (!_is_init) {
+ return Status::InternalError("Used before initialized.");
+ }
+
+ if (nullptr == block || nullptr == eos) {
+ return Status::InternalError("input pointer is nullptr.");
+ }
+
+ if (_tso_status_block == nullptr) {
+ RETURN_IF_ERROR(_get_tso_status_block_from_fe());
+ }
+
+ if (_row_idx == _total_rows) {
+ *eos = true;
+ return Status::OK();
+ }
+
+ int current_batch_rows = std::min(_block_rows_limit, _total_rows -
_row_idx);
+ ScopedMutableBlock scoped_mblock(block);
+ auto& mblock = scoped_mblock.mutable_block();
+ RETURN_IF_ERROR(mblock.add_rows(_tso_status_block.get(), _row_idx,
current_batch_rows));
+ _row_idx += current_batch_rows;
+
+ *eos = _row_idx == _total_rows;
+ return Status::OK();
+}
+
+} // namespace doris
diff --git a/be/src/information_schema/schema_tso_status_scanner.h
b/be/src/information_schema/schema_tso_status_scanner.h
new file mode 100644
index 00000000000..d405568e488
--- /dev/null
+++ b/be/src/information_schema/schema_tso_status_scanner.h
@@ -0,0 +1,54 @@
+// 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.
+
+#pragma once
+
+#include <memory>
+#include <vector>
+
+#include "common/status.h"
+#include "information_schema/schema_scanner.h"
+
+namespace doris {
+
+class RuntimeState;
+class Block;
+class TFetchSchemaTableDataResult;
+
+class SchemaTsoStatusScanner : public SchemaScanner {
+ ENABLE_FACTORY_CREATOR(SchemaTsoStatusScanner);
+
+public:
+ SchemaTsoStatusScanner();
+ ~SchemaTsoStatusScanner() override;
+
+ Status start(RuntimeState* state) override;
+ Status get_next_block_internal(Block* block, bool* eos) override;
+
+private:
+ Status _get_tso_status_block_from_fe();
+ Status _process_tso_status_result(const TFetchSchemaTableDataResult&
result);
+
+ int _block_rows_limit = 4096;
+ int _row_idx = 0;
+ int _total_rows = 0;
+ std::unique_ptr<Block> _tso_status_block = nullptr;
+ int _rpc_timeout_ms = 3000;
+ static std::vector<SchemaScanner::ColumnDesc> _s_tso_status_columns;
+};
+
+} // namespace doris
diff --git a/be/test/exec/schema_scanner/schema_tso_status_scanner_test.cpp
b/be/test/exec/schema_scanner/schema_tso_status_scanner_test.cpp
new file mode 100644
index 00000000000..67b82e2f181
--- /dev/null
+++ b/be/test/exec/schema_scanner/schema_tso_status_scanner_test.cpp
@@ -0,0 +1,228 @@
+// 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 "information_schema/schema_tso_status_scanner.h"
+
+#include <gen_cpp/FrontendService_types.h>
+#include <gtest/gtest.h>
+
+#include <array>
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "common/object_pool.h"
+#include "core/block/block.h"
+#include "runtime/runtime_state.h"
+#include "testutil/mock/mock_runtime_state.h"
+
+namespace doris {
+namespace {
+
+TRow create_tso_status_row(const std::array<int64_t, 4>& values) {
+ std::vector<TCell> cells;
+ cells.reserve(values.size());
+ for (int64_t value : values) {
+ TCell cell;
+ cell.__set_longVal(value);
+ cells.emplace_back(std::move(cell));
+ }
+
+ TRow row;
+ row.__set_column_value(cells);
+ return row;
+}
+
+TFetchSchemaTableDataResult create_tso_status_result(const std::vector<TRow>&
rows) {
+ TFetchSchemaTableDataResult result;
+ result.status.__set_status_code(TStatusCode::OK);
+ result.__set_data_batch(rows);
+ return result;
+}
+
+std::unique_ptr<Block> create_output_block(SchemaTsoStatusScanner* scanner) {
+ auto block = Block::create_unique();
+ scanner->_init_block(block.get());
+ return block;
+}
+
+void expect_tso_status_row(const Block& block, size_t row_idx,
+ const std::array<int64_t, 4>& expected) {
+ ASSERT_EQ(expected.size(), block.columns());
+ for (size_t column_idx = 0; column_idx < expected.size(); ++column_idx) {
+ const auto& column = block.get_by_position(column_idx).column;
+ EXPECT_FALSE(column->is_null_at(row_idx));
+ EXPECT_EQ(expected[column_idx], (*column)[row_idx].get<TYPE_BIGINT>());
+ }
+}
+
+} // namespace
+
+TEST(SchemaTsoStatusScannerTest, test_create_tso_status_scanner) {
+ auto scanner = SchemaScanner::create(TSchemaTableType::SCH_TSO_STATUS);
+ ASSERT_NE(nullptr, scanner);
+ EXPECT_EQ(TSchemaTableType::SCH_TSO_STATUS, scanner->type());
+ ASSERT_EQ(4, scanner->get_column_desc().size());
+ EXPECT_STREQ("WINDOW_END_PHYSICAL_TIME",
scanner->get_column_desc()[0].name);
+ EXPECT_STREQ("CURRENT_TSO", scanner->get_column_desc()[1].name);
+ EXPECT_STREQ("CURRENT_TSO_PHYSICAL_TIME",
scanner->get_column_desc()[2].name);
+ EXPECT_STREQ("CURRENT_TSO_LOGICAL_COUNTER",
scanner->get_column_desc()[3].name);
+ for (const auto& column : scanner->get_column_desc()) {
+ EXPECT_EQ(TYPE_BIGINT, column.type);
+ EXPECT_TRUE(column.is_null);
+ }
+}
+
+TEST(SchemaTsoStatusScannerTest, test_start) {
+ MockRuntimeState state;
+ state._batch_size = 2;
+ state._query_options.__set_execution_timeout(7);
+ SchemaScannerParam param;
+ ObjectPool pool;
+
+ SchemaTsoStatusScanner scanner;
+ ASSERT_TRUE(scanner.init(&state, ¶m, &pool).ok());
+ ASSERT_TRUE(scanner.start(&state).ok());
+
+ EXPECT_EQ(2, scanner._block_rows_limit);
+ EXPECT_EQ(7000, scanner._rpc_timeout_ms);
+}
+
+TEST(SchemaTsoStatusScannerTest, test_get_next_block_invalid_input) {
+ SchemaTsoStatusScanner scanner;
+ auto block = Block::create_unique();
+ bool eos = false;
+
+ auto status = scanner.get_next_block_internal(block.get(), &eos);
+ EXPECT_TRUE(status.is<ErrorCode::INTERNAL_ERROR>());
+ EXPECT_FALSE(eos);
+
+ MockRuntimeState state;
+ SchemaScannerParam param;
+ ObjectPool pool;
+ ASSERT_TRUE(scanner.init(&state, ¶m, &pool).ok());
+
+ status = scanner.get_next_block_internal(nullptr, &eos);
+ EXPECT_TRUE(status.is<ErrorCode::INTERNAL_ERROR>());
+ status = scanner.get_next_block_internal(block.get(), nullptr);
+ EXPECT_TRUE(status.is<ErrorCode::INTERNAL_ERROR>());
+}
+
+TEST(SchemaTsoStatusScannerTest, test_process_tso_status_result_error) {
+ TFetchSchemaTableDataResult result;
+ result.status.__set_status_code(TStatusCode::INTERNAL_ERROR);
+ result.status.__set_error_msgs({"fetch failed"});
+
+ SchemaTsoStatusScanner scanner;
+ auto status = scanner._process_tso_status_result(result);
+
+ EXPECT_TRUE(status.is<ErrorCode::INTERNAL_ERROR>());
+ EXPECT_NE(std::string::npos, status.to_string().find("fetch failed"));
+ EXPECT_EQ(nullptr, scanner._tso_status_block);
+}
+
+TEST(SchemaTsoStatusScannerTest, test_process_tso_status_result) {
+ const std::array<int64_t, 4> first_row = {1000, 2000, 3000, 4000};
+ const std::array<int64_t, 4> second_row = {1001, 2001, 3001, 4001};
+ auto result = create_tso_status_result(
+ {create_tso_status_row(first_row),
create_tso_status_row(second_row)});
+
+ SchemaTsoStatusScanner scanner;
+ ASSERT_TRUE(scanner._process_tso_status_result(result).ok());
+
+ ASSERT_NE(nullptr, scanner._tso_status_block);
+ EXPECT_EQ(4, scanner._tso_status_block->columns());
+ EXPECT_EQ(2, scanner._tso_status_block->rows());
+ EXPECT_EQ(2, scanner._total_rows);
+ expect_tso_status_row(*scanner._tso_status_block, 0, first_row);
+ expect_tso_status_row(*scanner._tso_status_block, 1, second_row);
+}
+
+TEST(SchemaTsoStatusScannerTest,
test_process_tso_status_result_schema_mismatch) {
+ TRow invalid_row = create_tso_status_row({1000, 2000, 3000, 4000});
+ invalid_row.column_value.pop_back();
+ auto result = create_tso_status_result({invalid_row});
+
+ SchemaTsoStatusScanner scanner;
+ auto status = scanner._process_tso_status_result(result);
+
+ EXPECT_TRUE(status.is<ErrorCode::INTERNAL_ERROR>());
+ EXPECT_NE(std::string::npos,
+ status.to_string().find("TSO status schema does not match
between FE and BE"));
+ EXPECT_EQ(0, scanner._total_rows);
+}
+
+TEST(SchemaTsoStatusScannerTest, test_get_next_block_empty_result) {
+ MockRuntimeState state;
+ SchemaScannerParam param;
+ ObjectPool pool;
+ SchemaTsoStatusScanner scanner;
+ ASSERT_TRUE(scanner.init(&state, ¶m, &pool).ok());
+ ASSERT_TRUE(scanner.start(&state).ok());
+
ASSERT_TRUE(scanner._process_tso_status_result(create_tso_status_result({})).ok());
+
+ auto block = create_output_block(&scanner);
+ bool eos = false;
+ ASSERT_TRUE(scanner.get_next_block_internal(block.get(), &eos).ok());
+
+ EXPECT_TRUE(eos);
+ EXPECT_EQ(0, block->rows());
+}
+
+TEST(SchemaTsoStatusScannerTest, test_get_next_block_in_batches) {
+ const std::array<int64_t, 4> first_row = {1000, 2000, 3000, 4000};
+ const std::array<int64_t, 4> second_row = {1001, 2001, 3001, 4001};
+ const std::array<int64_t, 4> third_row = {1002, 2002, 3002, 4002};
+
+ MockRuntimeState state;
+ state._batch_size = 2;
+ SchemaScannerParam param;
+ ObjectPool pool;
+ SchemaTsoStatusScanner scanner;
+ ASSERT_TRUE(scanner.init(&state, ¶m, &pool).ok());
+ ASSERT_TRUE(scanner.start(&state).ok());
+ ASSERT_TRUE(scanner._process_tso_status_result(
+
create_tso_status_result({create_tso_status_row(first_row),
+
create_tso_status_row(second_row),
+
create_tso_status_row(third_row)}))
+ .ok());
+
+ auto first_block = create_output_block(&scanner);
+ bool eos = true;
+ ASSERT_TRUE(scanner.get_next_block_internal(first_block.get(), &eos).ok());
+ EXPECT_FALSE(eos);
+ ASSERT_EQ(2, first_block->rows());
+ expect_tso_status_row(*first_block, 0, first_row);
+ expect_tso_status_row(*first_block, 1, second_row);
+
+ auto second_block = create_output_block(&scanner);
+ ASSERT_TRUE(scanner.get_next_block_internal(second_block.get(),
&eos).ok());
+ EXPECT_TRUE(eos);
+ ASSERT_EQ(1, second_block->rows());
+ expect_tso_status_row(*second_block, 0, third_row);
+
+ auto exhausted_block = create_output_block(&scanner);
+ eos = false;
+ ASSERT_TRUE(scanner.get_next_block_internal(exhausted_block.get(),
&eos).ok());
+ EXPECT_TRUE(eos);
+ EXPECT_EQ(0, exhausted_block->rows());
+}
+
+} // namespace doris
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/analysis/SchemaTableType.java
b/fe/fe-core/src/main/java/org/apache/doris/analysis/SchemaTableType.java
index 6cdcb2f5918..843bbf717f3 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/SchemaTableType.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/SchemaTableType.java
@@ -125,7 +125,8 @@ public enum SchemaTableType {
SCH_ROLE_MAPPINGS("ROLE_MAPPINGS", "ROLE_MAPPINGS",
TSchemaTableType.SCH_ROLE_MAPPINGS),
SCH_BACKEND_MS_RPC_TABLE_THROTTLERS("BACKEND_MS_RPC_TABLE_THROTTLERS",
"BACKEND_MS_RPC_TABLE_THROTTLERS",
TSchemaTableType.SCH_BACKEND_MS_RPC_TABLE_THROTTLERS),
- SCH_EXTENSIONS("EXTENSIONS", "EXTENSIONS",
TSchemaTableType.SCH_EXTENSIONS);
+ SCH_EXTENSIONS("EXTENSIONS", "EXTENSIONS",
TSchemaTableType.SCH_EXTENSIONS),
+ SCH_TSO_STATUS("TSO_STATUS", "TSO_STATUS",
TSchemaTableType.SCH_TSO_STATUS);
private static final String dbName = "INFORMATION_SCHEMA";
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/SchemaTable.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/SchemaTable.java
index 5bf68fcc5c8..0c6ecefd239 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/SchemaTable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/SchemaTable.java
@@ -897,6 +897,16 @@ public class SchemaTable extends Table {
.column("LAG",
ScalarType.createVarchar(NAME_CHAR_LEN))
.column("LAST_CONSUMPTION_TIME",
ScalarType.createType(PrimitiveType.BIGINT))
.build()))
+ .put("tso_status",
+ new SchemaTable(SystemIdGenerator.getNextId(),
"tso_status", TableType.SCHEMA,
+ builder().column("WINDOW_END_PHYSICAL_TIME",
+
ScalarType.createType(PrimitiveType.BIGINT))
+ .column("CURRENT_TSO",
ScalarType.createType(PrimitiveType.BIGINT))
+ .column("CURRENT_TSO_PHYSICAL_TIME",
+
ScalarType.createType(PrimitiveType.BIGINT))
+ .column("CURRENT_TSO_LOGICAL_COUNTER",
+
ScalarType.createType(PrimitiveType.BIGINT))
+ .build()))
.put("be_compaction_tasks",
new SchemaTable(SystemIdGenerator.getNextId(),
"be_compaction_tasks", TableType.SCHEMA,
builder().column("BACKEND_ID",
ScalarType.createType(PrimitiveType.BIGINT))
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/TSOAction.java
b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/TSOAction.java
index cbcd736e8da..1e7b57563d8 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/TSOAction.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/TSOAction.java
@@ -19,6 +19,7 @@ package org.apache.doris.httpv2.rest;
import org.apache.doris.catalog.Env;
import org.apache.doris.httpv2.entity.ResponseEntityBuilder;
+import org.apache.doris.tso.TSOService;
import org.apache.doris.tso.TSOTimestamp;
import com.google.common.collect.Maps;
@@ -38,6 +39,7 @@ import java.util.Map;
*
* Example usage:
* GET /api/tso
+ * GET /api/tso?local=true
* Response:
* {
* "window_end_physical_time": 1625097600000,
@@ -54,6 +56,8 @@ public class TSOAction extends RestBaseController {
/**
* Get current TSO information.
* This interface only returns TSO information without increasing the TSO
value.
+ * Requests are forwarded to the master FE by default. Set local=true to
query
+ * the TSO state of the FE that receives the request.
*
* @param request HTTP request
* @param response HTTP response
@@ -69,25 +73,24 @@ public class TSOAction extends RestBaseController {
//check user auth
executeCheckPassword(request, response);
+ if (!Boolean.parseBoolean(request.getParameter("local")) &&
checkForwardToMaster(request)) {
+ return forwardToMaster(request);
+ }
+
Env env = Env.getCurrentEnv();
if (env == null || !env.isReady()) {
LOG.warn("TSO HTTP API: FE is not ready");
return ResponseEntityBuilder.badRequest("FE is not ready");
}
- if (!env.isMaster()) {
- LOG.warn("TSO HTTP API: current FE is not master");
- return ResponseEntityBuilder.badRequest("Current FE is not
master");
- }
- // Get current TSO information without increasing it
- long windowEndPhysicalTime = env.getTSOService().getWindowEndTSO();
- long currentTSO = env.getTSOService().getCurrentTSO();
+ TSOService.TSOStatusSnapshot statusSnapshot =
env.getTSOService().getStatusSnapshot();
+ long currentTso = statusSnapshot.getCurrentTso();
// Prepare response data with detailed TSO information
Map<String, Object> result = Maps.newHashMap();
- result.put("window_end_physical_time", windowEndPhysicalTime);
- result.put("current_tso", currentTSO);
- result.put("current_tso_physical_time",
TSOTimestamp.extractPhysicalTime(currentTSO));
- result.put("current_tso_logical_counter",
TSOTimestamp.extractLogicalCounter(currentTSO));
+ result.put("window_end_physical_time",
statusSnapshot.getWindowEndPhysicalTime());
+ result.put("current_tso", currentTso);
+ result.put("current_tso_physical_time",
TSOTimestamp.extractPhysicalTime(currentTso));
+ result.put("current_tso_logical_counter",
TSOTimestamp.extractLogicalCounter(currentTso));
return ResponseEntityBuilder.ok(result);
} catch (Exception e) {
LOG.warn("Failed to get TSO information", e);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/MetadataGenerator.java
b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/MetadataGenerator.java
index d7908634392..19ac5631c93 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/MetadataGenerator.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/MetadataGenerator.java
@@ -46,6 +46,7 @@ import org.apache.doris.catalog.Type;
import org.apache.doris.catalog.View;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.ClientPool;
+import org.apache.doris.common.Config;
import org.apache.doris.common.ErrorCode;
import org.apache.doris.common.FeConstants;
import org.apache.doris.common.Pair;
@@ -116,6 +117,8 @@ import org.apache.doris.thrift.TStatusCode;
import org.apache.doris.thrift.TTasksMetadataParams;
import org.apache.doris.thrift.TUnit;
import org.apache.doris.thrift.TUserIdentity;
+import org.apache.doris.tso.TSOService;
+import org.apache.doris.tso.TSOTimestamp;
import com.codahale.metrics.Snapshot;
import com.google.common.base.Joiner;
@@ -175,6 +178,8 @@ public class MetadataGenerator {
private static final ImmutableMap<String, Integer>
TABLE_STREAM_CONSUMPTION_COLUMN_TO_INDEX;
+ private static final ImmutableMap<String, Integer>
TSO_STATUS_COLUMN_TO_INDEX;
+
static {
ImmutableMap.Builder<String, Integer> activeQueriesbuilder = new
ImmutableMap.Builder();
List<Column> activeQueriesColList =
SchemaTable.TABLE_MAP.get("active_queries").getFullSchema();
@@ -291,6 +296,13 @@ public class MetadataGenerator {
tableStreamConsumptionBuilder.put(tableStreamConsumptionBuilderColList.get(i).getName().toLowerCase(),
i);
}
TABLE_STREAM_CONSUMPTION_COLUMN_TO_INDEX =
tableStreamConsumptionBuilder.build();
+
+ ImmutableMap.Builder<String, Integer> tsoStatusBuilder = new
ImmutableMap.Builder();
+ List<Column> tsoStatusColList =
SchemaTable.TABLE_MAP.get("tso_status").getFullSchema();
+ for (int i = 0; i < tsoStatusColList.size(); i++) {
+
tsoStatusBuilder.put(tsoStatusColList.get(i).getName().toLowerCase(), i);
+ }
+ TSO_STATUS_COLUMN_TO_INDEX = tsoStatusBuilder.build();
}
public static TFetchSchemaTableDataResult
getMetadataTable(TFetchSchemaTableDataRequest request) throws TException {
@@ -422,6 +434,10 @@ public class MetadataGenerator {
result = streamConsumptionMetadataResult(schemaTableParams);
columnIndex = TABLE_STREAM_CONSUMPTION_COLUMN_TO_INDEX;
break;
+ case TSO_STATUS:
+ result = tsoStatusMetadataResult();
+ columnIndex = TSO_STATUS_COLUMN_TO_INDEX;
+ break;
default:
return errorResult("invalid schema table name.");
}
@@ -2169,4 +2185,27 @@ public class MetadataGenerator {
result.setStatus(new TStatus(TStatusCode.OK));
return result;
}
+
+ private static TFetchSchemaTableDataResult tsoStatusMetadataResult() {
+ if (!Config.enable_feature_binlog) {
+ return errorResult("TSO feature is disabled, please check
enable_feature_binlog");
+ }
+
+ TSOService.TSOStatusSnapshot statusSnapshot =
Env.getCurrentEnv().getTSOService().getStatusSnapshot();
+ if (!statusSnapshot.isInitialized()) {
+ return errorResult("TSO timestamp is not calibrated, please
check");
+ }
+
+ long currentTso = statusSnapshot.getCurrentTso();
+ TRow row = new TRow();
+ row.addToColumnValue(new
TCell().setLongVal(statusSnapshot.getWindowEndPhysicalTime()));
+ row.addToColumnValue(new TCell().setLongVal(currentTso));
+ row.addToColumnValue(new
TCell().setLongVal(TSOTimestamp.extractPhysicalTime(currentTso)));
+ row.addToColumnValue(new
TCell().setLongVal(TSOTimestamp.extractLogicalCounter(currentTso)));
+
+ TFetchSchemaTableDataResult result = new TFetchSchemaTableDataResult();
+ result.setDataBatch(Lists.newArrayList(row));
+ result.setStatus(new TStatus(TStatusCode.OK));
+ return result;
+ }
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/tso/TSOService.java
b/fe/fe-core/src/main/java/org/apache/doris/tso/TSOService.java
index ffd9bb536d8..2acfff15564 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/tso/TSOService.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/tso/TSOService.java
@@ -49,6 +49,33 @@ public class TSOService extends MasterDaemon {
private final AtomicBoolean fatalClockBackwardReported = new
AtomicBoolean(false);
private final AtomicLong windowEndTSO = new AtomicLong(0);
+ /**
+ * Immutable snapshot of the current TSO service status.
+ */
+ public static final class TSOStatusSnapshot {
+ private final boolean initialized;
+ private final long currentTso;
+ private final long windowEndPhysicalTime;
+
+ public TSOStatusSnapshot(boolean initialized, long currentTso, long
windowEndPhysicalTime) {
+ this.initialized = initialized;
+ this.currentTso = currentTso;
+ this.windowEndPhysicalTime = windowEndPhysicalTime;
+ }
+
+ public boolean isInitialized() {
+ return initialized;
+ }
+
+ public long getCurrentTso() {
+ return currentTso;
+ }
+
+ public long getWindowEndPhysicalTime() {
+ return windowEndPhysicalTime;
+ }
+ }
+
private static final class TSOClockBackwardException extends
RuntimeException {
private TSOClockBackwardException(String message) {
super(message);
@@ -235,6 +262,21 @@ public class TSOService extends MasterDaemon {
}
}
+ /**
+ * Get a read-only snapshot of the TSO service status without allocating a
new TSO timestamp.
+ *
+ * @return Current initialization state, composed TSO, and window end
physical time
+ */
+ public TSOStatusSnapshot getStatusSnapshot() {
+ lock.lock();
+ try {
+ return new TSOStatusSnapshot(
+ isInitialized.get(), globalTimestamp.composeTimestamp(),
windowEndTSO.get());
+ } finally {
+ lock.unlock();
+ }
+ }
+
/**
* Calibrate the TSO timestamp when service starts
* This ensures the timestamp is consistent with the last persisted value
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/catalog/SchemaTableTest.java
b/fe/fe-core/src/test/java/org/apache/doris/catalog/SchemaTableTest.java
index 2d3b1fe38fe..f7422064608 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/catalog/SchemaTableTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/catalog/SchemaTableTest.java
@@ -116,5 +116,14 @@ public class SchemaTableTest {
Assertions.assertEquals("CREATE_TIME",
roleMappings.getFullSchema().get(5).getName());
Assertions.assertEquals("ALTER_USER",
roleMappings.getFullSchema().get(6).getName());
Assertions.assertEquals("MODIFY_TIME",
roleMappings.getFullSchema().get(7).getName());
+
+ SchemaTable tsoStatus = (SchemaTable)
SchemaTable.TABLE_MAP.get("tso_status");
+ Assertions.assertFalse(tsoStatus.shouldFetchAllFe());
+ Assertions.assertFalse(tsoStatus.shouldAddAgg());
+ Assertions.assertEquals(4, tsoStatus.getFullSchema().size());
+ Assertions.assertEquals("WINDOW_END_PHYSICAL_TIME",
tsoStatus.getFullSchema().get(0).getName());
+ Assertions.assertEquals("CURRENT_TSO",
tsoStatus.getFullSchema().get(1).getName());
+ Assertions.assertEquals("CURRENT_TSO_PHYSICAL_TIME",
tsoStatus.getFullSchema().get(2).getName());
+ Assertions.assertEquals("CURRENT_TSO_LOGICAL_COUNTER",
tsoStatus.getFullSchema().get(3).getName());
}
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/httpv2/rest/TSOActionTest.java
b/fe/fe-core/src/test/java/org/apache/doris/httpv2/rest/TSOActionTest.java
new file mode 100644
index 00000000000..ad5208e542c
--- /dev/null
+++ b/fe/fe-core/src/test/java/org/apache/doris/httpv2/rest/TSOActionTest.java
@@ -0,0 +1,90 @@
+// 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.
+
+package org.apache.doris.httpv2.rest;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.httpv2.entity.ResponseBody;
+import org.apache.doris.tso.TSOService;
+import org.apache.doris.tso.TSOTimestamp;
+
+import jakarta.servlet.http.HttpServletRequest;
+import jakarta.servlet.http.HttpServletResponse;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+import org.springframework.http.ResponseEntity;
+
+import java.util.Map;
+
+public class TSOActionTest {
+ @Test
+ public void testDefaultRequestForwardsToMaster() {
+ HttpServletRequest request = Mockito.mock(HttpServletRequest.class);
+ HttpServletResponse response = Mockito.mock(HttpServletResponse.class);
+ TSOAction action = Mockito.spy(new TSOAction());
+ Object forwardedResult = new Object();
+
+ Mockito.doReturn(null).when(action).executeCheckPassword(request,
response);
+ Mockito.doReturn(true).when(action).checkForwardToMaster(request);
+
Mockito.doReturn(forwardedResult).when(action).forwardToMaster(request);
+
+ Assertions.assertSame(forwardedResult, action.getTSO(request,
response));
+ Mockito.verify(action).forwardToMaster(request);
+ }
+
+ @Test
+ public void testLocalRequestReadsLocalSnapshot() {
+ HttpServletRequest request = Mockito.mock(HttpServletRequest.class);
+ HttpServletResponse response = Mockito.mock(HttpServletResponse.class);
+ TSOAction action = Mockito.spy(new TSOAction());
+ Env env = Mockito.mock(Env.class);
+ TSOService tsoService = Mockito.mock(TSOService.class);
+ long physicalTime = 1_725_000_000_000L;
+ long logicalCounter = 17L;
+ long currentTso = TSOTimestamp.composeTimestamp(physicalTime,
logicalCounter);
+ long windowEndPhysicalTime = physicalTime + 5_000L;
+
+ Mockito.when(request.getParameter("local")).thenReturn("true");
+ Mockito.doReturn(null).when(action).executeCheckPassword(request,
response);
+ Mockito.when(env.isReady()).thenReturn(true);
+ Mockito.when(env.getTSOService()).thenReturn(tsoService);
+ Mockito.when(tsoService.getStatusSnapshot()).thenReturn(
+ new TSOService.TSOStatusSnapshot(true, currentTso,
windowEndPhysicalTime));
+
+ try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+
+ ResponseEntity<?> responseEntity = (ResponseEntity<?>)
action.getTSO(request, response);
+ ResponseBody<?> responseBody = (ResponseBody<?>)
responseEntity.getBody();
+ Map<?, ?> data = (Map<?, ?>) responseBody.getData();
+
+ Assertions.assertEquals(windowEndPhysicalTime,
data.get("window_end_physical_time"));
+ Assertions.assertEquals(currentTso, data.get("current_tso"));
+ Assertions.assertEquals(physicalTime,
data.get("current_tso_physical_time"));
+ Assertions.assertEquals(logicalCounter,
data.get("current_tso_logical_counter"));
+ }
+
+ Mockito.verify(action, Mockito.never()).checkForwardToMaster(request);
+ Mockito.verify(action, Mockito.never()).forwardToMaster(request);
+ Mockito.verify(tsoService).getStatusSnapshot();
+ Mockito.verify(tsoService, Mockito.never()).getTSO();
+ Mockito.verify(tsoService, Mockito.never()).getCurrentTSO();
+ Mockito.verify(tsoService, Mockito.never()).getWindowEndTSO();
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/tablefunction/TsoStatusMetadataGeneratorTest.java
b/fe/fe-core/src/test/java/org/apache/doris/tablefunction/TsoStatusMetadataGeneratorTest.java
new file mode 100644
index 00000000000..dc8761c5ac3
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/tablefunction/TsoStatusMetadataGeneratorTest.java
@@ -0,0 +1,132 @@
+// 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.
+
+package org.apache.doris.tablefunction;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.Config;
+import org.apache.doris.thrift.TFetchSchemaTableDataRequest;
+import org.apache.doris.thrift.TFetchSchemaTableDataResult;
+import org.apache.doris.thrift.TRow;
+import org.apache.doris.thrift.TSchemaTableName;
+import org.apache.doris.thrift.TSchemaTableRequestParams;
+import org.apache.doris.thrift.TStatusCode;
+import org.apache.doris.tso.TSOService;
+import org.apache.doris.tso.TSOTimestamp;
+
+import com.google.common.collect.ImmutableList;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+public class TsoStatusMetadataGeneratorTest {
+ private boolean originalEnableFeatureBinlog;
+ private Env env;
+ private TSOService tsoService;
+ private MockedStatic<Env> mockedEnv;
+
+ @BeforeEach
+ public void setUp() {
+ originalEnableFeatureBinlog = Config.enable_feature_binlog;
+ Config.enable_feature_binlog = true;
+
+ env = Mockito.mock(Env.class);
+ tsoService = Mockito.mock(TSOService.class);
+ mockedEnv = Mockito.mockStatic(Env.class);
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+ Mockito.when(env.getTSOService()).thenReturn(tsoService);
+ }
+
+ @AfterEach
+ public void tearDown() {
+ mockedEnv.close();
+ Config.enable_feature_binlog = originalEnableFeatureBinlog;
+ }
+
+ @Test
+ public void testTsoStatusResult() throws Exception {
+ long physicalTime = 1_725_000_000_000L;
+ long logicalCounter = 17L;
+ long currentTso = TSOTimestamp.composeTimestamp(physicalTime,
logicalCounter);
+ long windowEndPhysicalTime = physicalTime + 5_000L;
+ Mockito.when(tsoService.getStatusSnapshot()).thenReturn(
+ new TSOService.TSOStatusSnapshot(true, currentTso,
windowEndPhysicalTime));
+
+ TFetchSchemaTableDataResult result =
MetadataGenerator.getSchemaTableData(newRequest());
+
+ Assertions.assertEquals(TStatusCode.OK,
result.getStatus().getStatusCode());
+ Assertions.assertEquals(1, result.getDataBatchSize());
+ TRow row = result.getDataBatch().get(0);
+ Assertions.assertEquals(windowEndPhysicalTime,
row.getColumnValue().get(0).getLongVal());
+ Assertions.assertEquals(currentTso,
row.getColumnValue().get(1).getLongVal());
+ Assertions.assertEquals(physicalTime,
row.getColumnValue().get(2).getLongVal());
+ Assertions.assertEquals(logicalCounter,
row.getColumnValue().get(3).getLongVal());
+ Mockito.verify(tsoService).getStatusSnapshot();
+ Mockito.verify(tsoService, Mockito.never()).getTSO();
+ }
+
+ @Test
+ public void testColumnFiltering() throws Exception {
+ long physicalTime = 1_725_000_000_000L;
+ long currentTso = TSOTimestamp.composeTimestamp(physicalTime, 17L);
+ Mockito.when(tsoService.getStatusSnapshot()).thenReturn(
+ new TSOService.TSOStatusSnapshot(true, currentTso,
physicalTime + 5_000L));
+ TFetchSchemaTableDataRequest request = newRequest();
+ request.getSchemaTableParams().setColumnsName(
+ ImmutableList.of("current_tso_physical_time", "current_tso"));
+
+ TFetchSchemaTableDataResult result =
MetadataGenerator.getSchemaTableData(request);
+
+ Assertions.assertEquals(TStatusCode.OK,
result.getStatus().getStatusCode());
+ Assertions.assertEquals(2,
result.getDataBatch().get(0).getColumnValueSize());
+ Assertions.assertEquals(physicalTime,
result.getDataBatch().get(0).getColumnValue().get(0).getLongVal());
+ Assertions.assertEquals(currentTso,
result.getDataBatch().get(0).getColumnValue().get(1).getLongVal());
+ }
+
+ @Test
+ public void testDisabled() throws Exception {
+ Config.enable_feature_binlog = false;
+
+ TFetchSchemaTableDataResult result =
MetadataGenerator.getSchemaTableData(newRequest());
+
+ Assertions.assertEquals(TStatusCode.INTERNAL_ERROR,
result.getStatus().getStatusCode());
+
Assertions.assertTrue(result.getStatus().getErrorMsgs().get(0).contains("enable_feature_binlog"));
+ Mockito.verifyNoInteractions(tsoService);
+ }
+
+ @Test
+ public void testNotCalibrated() throws Exception {
+ Mockito.when(tsoService.getStatusSnapshot()).thenReturn(
+ new TSOService.TSOStatusSnapshot(false, 0L, 0L));
+
+ TFetchSchemaTableDataResult result =
MetadataGenerator.getSchemaTableData(newRequest());
+
+ Assertions.assertEquals(TStatusCode.INTERNAL_ERROR,
result.getStatus().getStatusCode());
+
Assertions.assertTrue(result.getStatus().getErrorMsgs().get(0).contains("not
calibrated"));
+ Mockito.verify(tsoService).getStatusSnapshot();
+ }
+
+ private TFetchSchemaTableDataRequest newRequest() {
+ TFetchSchemaTableDataRequest request = new
TFetchSchemaTableDataRequest();
+ request.setSchemaTableName(TSchemaTableName.TSO_STATUS);
+ request.setSchemaTableParams(new TSchemaTableRequestParams());
+ return request;
+ }
+}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/tso/TSOServiceTest.java
b/fe/fe-core/src/test/java/org/apache/doris/tso/TSOServiceTest.java
index 59740952e07..30a27bfc4f6 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/tso/TSOServiceTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/tso/TSOServiceTest.java
@@ -350,7 +350,10 @@ public class TSOServiceTest {
Assert.assertTrue(e.getMessage().contains("EditLog is null"));
}
- Assert.assertEquals(0L, tsoService.getWindowEndTSO());
+ TSOService.TSOStatusSnapshot statusSnapshot =
tsoService.getStatusSnapshot();
+ Assert.assertFalse(statusSnapshot.isInitialized());
+ Assert.assertTrue(statusSnapshot.getCurrentTso() > 0L);
+ Assert.assertEquals(0L, statusSnapshot.getWindowEndPhysicalTime());
try {
tsoService.getTSO();
diff --git a/gensrc/thrift/Descriptors.thrift b/gensrc/thrift/Descriptors.thrift
index 2b4fdd5b20d..302c57b3180 100644
--- a/gensrc/thrift/Descriptors.thrift
+++ b/gensrc/thrift/Descriptors.thrift
@@ -221,6 +221,7 @@ enum TSchemaTableType {
SCH_ROLE_MAPPINGS = 71;
SCH_BACKEND_MS_RPC_TABLE_THROTTLERS = 72;
SCH_EXTENSIONS = 73;
+ SCH_TSO_STATUS = 74;
}
enum THdfsCompression {
diff --git a/gensrc/thrift/FrontendService.thrift
b/gensrc/thrift/FrontendService.thrift
index 7774194a8eb..0d0fef3eacc 100644
--- a/gensrc/thrift/FrontendService.thrift
+++ b/gensrc/thrift/FrontendService.thrift
@@ -918,6 +918,7 @@ enum TSchemaTableName {
TABLE_STREAM_CONSUMPTION = 16,
ROLE_MAPPINGS = 17,
EXTENSIONS = 18,
+ TSO_STATUS = 19,
}
struct TMetadataTableRequestParams {
diff --git a/regression-test/suites/tso_p0/test_tso_api.groovy
b/regression-test/suites/tso_p0/test_tso_api.groovy
index fb376d64315..0f5a747ace4 100644
--- a/regression-test/suites/tso_p0/test_tso_api.groovy
+++ b/regression-test/suites/tso_p0/test_tso_api.groovy
@@ -43,6 +43,39 @@ suite("test_tso_api", "nonConcurrent") {
assertTrue(data.current_tso_physical_time > 0)
assertTrue(data.current_tso_logical_counter >= 0)
+ // Test local TSO API access
+ def localResult = Http.GET("${url}?local=true", true, true)
+ assertTrue(localResult.code == 0)
+ assertEquals(localResult.msg, "success")
+ assertTrue(localResult.data.containsKey("window_end_physical_time"))
+ assertTrue(localResult.data.containsKey("current_tso"))
+ assertTrue(localResult.data.containsKey("current_tso_physical_time"))
+ assertTrue(localResult.data.containsKey("current_tso_logical_counter"))
+
+ // Test information_schema TSO status interface
+ def statusRows = sql_return_maparray """
+ SELECT WINDOW_END_PHYSICAL_TIME, CURRENT_TSO,
+ CURRENT_TSO_PHYSICAL_TIME, CURRENT_TSO_LOGICAL_COUNTER
+ FROM information_schema.tso_status
+ """
+ assertEquals(1, statusRows.size())
+
+ def status = statusRows[0]
+ assertTrue(status.containsKey("WINDOW_END_PHYSICAL_TIME"))
+ assertTrue(status.containsKey("CURRENT_TSO"))
+ assertTrue(status.containsKey("CURRENT_TSO_PHYSICAL_TIME"))
+ assertTrue(status.containsKey("CURRENT_TSO_LOGICAL_COUNTER"))
+
+ long statusWindowEnd =
Long.parseLong(status.WINDOW_END_PHYSICAL_TIME.toString())
+ long statusCurrentTso = Long.parseLong(status.CURRENT_TSO.toString())
+ long statusPhysicalTime =
Long.parseLong(status.CURRENT_TSO_PHYSICAL_TIME.toString())
+ long statusLogicalCounter =
Long.parseLong(status.CURRENT_TSO_LOGICAL_COUNTER.toString())
+ assertTrue(statusWindowEnd >= statusPhysicalTime)
+ assertTrue(statusCurrentTso > 0)
+ assertTrue(statusPhysicalTime > 0)
+ assertTrue(statusLogicalCounter >= 0)
+ assertEquals(statusCurrentTso, (statusPhysicalTime << 18) |
statusLogicalCounter)
+
// Test 2: Multiple TSO API calls should return consistent increasing
values
def result1 = Http.GET(url, true, true)
Thread.sleep(10) // Small delay to ensure time progression
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
