This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-incremental-computation
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-incremental-computation
by this push:
new 4c321761520 branch-incremental-computation: pick the merged
incremental-computation PRs from master in merge order (#67783 #68034 #68033
#68057 #68094 #68087) (#68151)
4c321761520 is described below
commit 4c3217615207309c8c63c64b2919466f3019b3b2
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Fri Sep 18 10:55:09 2026 +0800
branch-incremental-computation: pick the merged incremental-computation PRs
from master in merge order (#67783 #68034 #68033 #68057 #68094 #68087) (#68151)
Cherry-picked from #67783, #68034, #68033, #68057, #68094, #68087
Batch pick of every merged PR carrying the `incremental-computation`
label that `branch-incremental-computation` does not have yet (no
`incremental-computation-picked` label), in the order they landed on
master (`git log --first-parent`). One commit per PR, each created with
`git cherry-pick -x` so the message ends with `(cherry picked from
commit <master sha>)`. Follows the same convention as #67830, #68017 and
#68073.
| # | Master commit | PR | Title |
|---|---|---|---|
| 1 | e545b13e089 | #67783 | [fix](policy) Enforce row policies on MOW
time travel |
| 2 | f7a0842f968 | #68034 | [fix](binlog) Fix missing DELETE events in
row binlog |
| 3 | e85575e3ad6 | #68033 | [fix](binlog) Preserve row binlog
compaction policy |
| 4 | 21160d78b42 | #68057 | [fix](binlog) Decouple row binlog
compaction from CCR binlog config |
| 5 | 04aa5a11584 | #68094 | [fix](binlog) Persist row binlog config
updates in cloud mode |
| 6 | da8feed859d | #68087 | [fix](binlog) Report streams with missing
base tables as stale |
Not included on purpose:
- The 20 labelled PRs that already carry
`incremental-computation-picked` (#62606 in the fork point, #67508 via
#67712, the nine of #67830, the six of #68017, the two of #68073, and
#68050 whose content this branch got directly through #68012).
- #68012 carries the label but is a PR against this branch itself
(merged as `6f7c87fa892`); nothing to pick.
- #67820 is still open on master; this branch already carries its
content via #67861.
### Prerequisite check
For every pick I listed the master commits between the fork point
(`efedf10c7e3`) and the pick that touch the same files and are not on
this branch, and checked whether the pick's behavior depends on them.
- **#67783** declares no related PR. It builds on the MOW time-travel
rewrite (`BindRelation.buildMowTimeTravelUnion`, #67480) which is
already here. Two unlabelled master commits overlap: #66770
(authorization plugin SPI: reworks the data-mask / row-filter API of
`LogicalCheckPolicy` and its test) and #67811 (removes the
`isPlayNereidsDump()` check in `CheckPolicy`). Neither is a functional
prerequisite: the fix consists of `getPolicyTable()` unwrapping
`OlapTableWrapper`, `CheckPolicy` collecting the whole filter chain
below the policy mask, and `BindRelation` putting a `LogicalCheckPolicy`
above each union branch — none of that uses the #66770 API. Not picked;
adapted instead (see below).
- **#68034**, **#68033**, **#68087** (#68087 relates to #67173, which is
before the fork point): no unpicked master commit touches any of their
files; the picks applied cleanly and are byte-identical to master.
- **#68057**: only `be/test/cloud/cloud_compaction_test.cpp` overlaps
with the unrelated #67972 (refresh tablet meta of continuously ingested
tablets); auto-merged, the pick only adds new `TEST_F` blocks. Main-code
hunks are byte-identical to master.
- **#68094**: overlaps with #68090 / #67972 / #66598
(`cloud_tablet.cpp`), #67295 / #67618 / #68090 (`base_tablet.{h,cpp}`),
#66598 / #67637 (`meta_service.cpp`), #66598 (`cloud.proto`,
`cloud_tablet_test.cpp`) and #67761 / #66598 / #67637
(`meta_service_test.cpp`). All auto-merged. The fix —
`BaseTablet::binlog_config()` under the meta lock,
`BinlogConfig::operator==`, `CloudTablet::sync_meta()` refreshing the
binlog config, `update_tablet` in the meta service accepting
`binlog_config`, and `CloudSchemaChangeHandler` routing ROW-binlog
property updates through it — does not use anything those commits add.
Its hunks are byte-identical to master except one trailing context line
in `sync_meta()` (`last_sync_tablet_meta_time_s` comes from #67972).
Only #67783 needed adaptation, recorded in its commit message:
- `LogicalCheckPolicy.java`: the conflicting context was master's
`parsePolicyExpression()` helper (#66770), which does not exist here;
the new `getPolicyTable()` is inserted in the same place without it.
- `CheckRowPolicyTest.java`: the data-mask mock is written against this
branch's per-column `AccessControllerManager.evalDataMaskPolicy(...,
column) -> Optional<DataMaskPolicy>` instead of master's
`evalDataMaskPolicies(..., Set<String>) -> Map<String, DataMaskSpec>`.
The masks are the same (concat for the random-distribution table; the
non-movable `k2` mask plus identity masks for the hidden reconstruction
columns of the MOW table). The master-only `Or` import (#66770) is not
carried; `Collections` / `Locale` imports were added because they arrive
with #66770 on master.
### Drift check against master
Each pick's `+`/`-` lines are identical to the master commit's, except
for the #67783 adaptation above. After the six picks, the touched files
still differ from master at `da8feed859d` in: `cloud_tablet.cpp`,
`base_tablet.{h,cpp}`, `cloud_compaction_test.cpp`,
`cloud_tablet_test.cpp`, `meta_service.cpp`, `meta_service_test.cpp`,
`cloud.proto`, `CheckPolicy.java`, `LogicalCheckPolicy.java`,
`CheckRowPolicyTest.java`. Replaying the nine unpicked master commits
listed above (#67637, #67618, #66770, #66598, #67972, #67811, #67761,
#67295, #68090) in a temporary index and removing this branch's own
#67861 (`GetTsoRecoveryTransactions` in `cloud.proto` /
`meta_service_test.cpp`) brings every file to zero diff against master,
except the two `LogicalCheckPolicy.java` / `CheckRowPolicyTest.java`
hunks of #66770 that overlap the adaptation. Nothing else is left over.
### Verification
- FE: `run-fe-ut.sh --run` on this branch (regenerates thrift/protobuf,
compiles fe-core main + test) with the test classes touched by the picks
plus `DeleteFromCommandTest` from #68034's checklist: 5 classes, 58
tests, 0 failures, 0 errors, BUILD SUCCESS — `CheckRowPolicyTest` 9 (the
three tests #67783 adds included), `ExplainTableStreamPlanTest` 24,
`CloudSchemaChangeHandlerTest` 18 (the tests of #68033 and #68094
included), `DropTableStreamTest` 5, `DeleteFromCommandTest` 2.
- FE checkstyle on fe-core: 0 violations.
- BE: `-fsyntax-only` with the flags of the Release build
(`compile_commands.json`, regenerated `gen_cpp` headers incl. the new
`TabletMetaInfoPB.binlog_config`) passes for `cloud/cloud_tablet.cpp`,
`cloud/cloud_storage_engine.cpp`, `storage/olap_server.cpp`,
`storage/tablet/tablet.cpp`, `storage/tablet/base_tablet.cpp`, and with
`-DBE_TEST -fno-access-control` for
`test/cloud/cloud_compaction_test.cpp`,
`test/cloud/cloud_tablet_test.cpp`,
`test/storage/compaction/compaction_task_test.cpp`,
`test/storage/tablet/tablet_test.cpp`.
- Meta service: the two-line `update_tablet` hunk of #68094 only uses
the generated `TabletMetaInfoPB::has_binlog_config()/binlog_config()`
and `TabletMetaCloudPB::mutable_binlog_config()` accessors, all present
in the regenerated headers.
- The two new groovy suites
(`row_binlog_p0/test_row_binlog_mow_light_delete`,
`time_travel_p0/test_mow_time_travel_row_policy`) parse cleanly (groovy
parser check).
---------
Co-authored-by: morrySnow <[email protected]>
Co-authored-by: Luwei <[email protected]>
---
be/src/cloud/cloud_storage_engine.cpp | 2 +-
be/src/cloud/cloud_tablet.cpp | 10 +-
be/src/storage/binlog_config.h | 2 +
be/src/storage/olap_server.cpp | 2 +-
be/src/storage/tablet/base_tablet.cpp | 4 +-
be/src/storage/tablet/base_tablet.h | 5 +
be/src/storage/tablet/tablet.cpp | 3 +-
be/test/cloud/cloud_compaction_test.cpp | 50 ++++++
be/test/cloud/cloud_tablet_test.cpp | 30 ++++
.../storage/compaction/compaction_task_test.cpp | 40 +++++
be/test/storage/tablet/tablet_test.cpp | 26 +++
cloud/src/meta-service/meta_service.cpp | 2 +
cloud/test/meta_service_test.cpp | 34 ++++
.../doris/catalog/stream/TableStreamManager.java | 12 +-
.../cloud/alter/CloudSchemaChangeHandler.java | 22 ++-
.../java/org/apache/doris/load/DeleteHandler.java | 7 +-
.../doris/nereids/rules/analysis/BindRelation.java | 44 ++---
.../doris/nereids/rules/analysis/CheckPolicy.java | 21 +--
.../trees/plans/commands/DeleteFromCommand.java | 5 +-
.../trees/plans/logical/LogicalCheckPolicy.java | 19 ++-
.../apache/doris/catalog/DropTableStreamTest.java | 69 ++++++++
.../cloud/alter/CloudSchemaChangeHandlerTest.java | 128 ++++++++++++++
.../nereids/rules/analysis/CheckRowPolicyTest.java | 183 +++++++++++++++++++--
.../trees/plans/ExplainTableStreamPlanTest.java | 24 +++
gensrc/proto/cloud.proto | 1 +
.../test_row_binlog_mow_light_delete.out | 16 ++
.../test_mow_time_travel_row_policy.out | 15 ++
.../test_row_binlog_mow_light_delete.groovy | 104 ++++++++++++
.../test_mow_time_travel_row_policy.groovy | 107 ++++++++++++
29 files changed, 913 insertions(+), 74 deletions(-)
diff --git a/be/src/cloud/cloud_storage_engine.cpp
b/be/src/cloud/cloud_storage_engine.cpp
index 6a2d46469d8..b9d88260ee4 100644
--- a/be/src/cloud/cloud_storage_engine.cpp
+++ b/be/src/cloud/cloud_storage_engine.cpp
@@ -681,7 +681,7 @@ void
CloudStorageEngine::_binlog_compaction_tasks_producer_callback() {
int64_t interval = config::generate_compaction_tasks_interval_ms;
do {
int64_t cur_time = UnixMillis();
- if (config::enable_feature_binlog && !config::disable_auto_compaction)
{
+ if (!config::disable_auto_compaction) {
Status st = _adjust_compaction_thread_num();
if (!st.ok()) {
break;
diff --git a/be/src/cloud/cloud_tablet.cpp b/be/src/cloud/cloud_tablet.cpp
index feb6f514668..542a33b2278 100644
--- a/be/src/cloud/cloud_tablet.cpp
+++ b/be/src/cloud/cloud_tablet.cpp
@@ -846,8 +846,7 @@ Result<std::unique_ptr<RowsetWriter>>
CloudTablet::create_rowset_writer(
context.enable_unique_key_merge_on_write =
enable_unique_key_merge_on_write();
context.encrypt_algorithm = tablet_meta()->encryption_algorithm();
if (context.write_binlog_opt().enable) {
- context.write_binlog_opt().set_need_before(
- tablet_meta()->binlog_config().need_historical_value());
+
context.write_binlog_opt().set_need_before(binlog_config().need_historical_value());
}
context.inverted_index_storage_format =
tablet_meta()->inverted_index_storage_format();
context.persist_inverted_index_storage_format =
@@ -891,8 +890,7 @@ Result<std::unique_ptr<RowsetWriter>>
CloudTablet::create_transient_rowset_write
context.is_transient_rowset_writer = true;
if (rowset.rowset_meta() != nullptr &&
rowset.rowset_meta()->is_row_binlog()) {
context.write_binlog_opt().enable = true;
- context.write_binlog_opt().set_need_before(
- tablet_meta()->binlog_config().need_historical_value());
+
context.write_binlog_opt().set_need_before(binlog_config().need_historical_value());
}
context.rowset_id = rowset.rowset_id();
context.tablet_id = tablet_id();
@@ -1554,6 +1552,7 @@ Status CloudTablet::sync_meta() {
auto new_disable_auto_compaction =
tablet_meta->tablet_schema()->disable_auto_compaction();
auto new_vertical_compaction_num_columns_per_group =
tablet_meta->vertical_compaction_num_columns_per_group();
+ auto new_binlog_config = tablet_meta->binlog_config();
{
std::unique_lock wlock(_meta_lock);
@@ -1598,6 +1597,9 @@ Status CloudTablet::sync_meta() {
_tablet_meta->set_vertical_compaction_num_columns_per_group(
new_vertical_compaction_num_columns_per_group);
}
+ if (_tablet_meta->binlog_config() != new_binlog_config) {
+ _tablet_meta->set_binlog_config(new_binlog_config);
+ }
}
return Status::OK();
diff --git a/be/src/storage/binlog_config.h b/be/src/storage/binlog_config.h
index 696eb696322..76751389a43 100644
--- a/be/src/storage/binlog_config.h
+++ b/be/src/storage/binlog_config.h
@@ -46,6 +46,8 @@ public:
BinlogConfig& operator=(BinlogConfig&&) = default;
~BinlogConfig() = default;
+ bool operator==(const BinlogConfig&) const = default;
+
bool is_enable() const { return _enable; }
void set_enable(bool enable) { _enable = enable; }
diff --git a/be/src/storage/olap_server.cpp b/be/src/storage/olap_server.cpp
index 2dc715cd327..7bcf615cf04 100644
--- a/be/src/storage/olap_server.cpp
+++ b/be/src/storage/olap_server.cpp
@@ -776,7 +776,7 @@ void
StorageEngine::_binlog_compaction_tasks_producer_callback() {
int64_t interval = config::generate_compaction_tasks_interval_ms;
do {
int64_t cur_time = UnixMillis();
- if (config::enable_feature_binlog && !config::disable_auto_compaction
&&
+ if (!config::disable_auto_compaction &&
(!config::enable_compaction_pause_on_high_memory ||
!GlobalMemoryArbitrator::is_exceed_soft_mem_limit(GB_EXCHANGE_BYTE))) {
_adjust_compaction_thread_num();
diff --git a/be/src/storage/tablet/base_tablet.cpp
b/be/src/storage/tablet/base_tablet.cpp
index c4a487f80ce..f9dc3d3ed4a 100644
--- a/be/src/storage/tablet/base_tablet.cpp
+++ b/be/src/storage/tablet/base_tablet.cpp
@@ -1516,9 +1516,9 @@ Status BaseTablet::update_delete_bitmap(const
BaseTabletSPtr& self, TabletTxnInf
binlog_rs->rowset_meta()->is_row_binlog()) {
DCHECK(txn_info->attach_row_binlog.tablet != nullptr);
row_binlog_rowset = binlog_rs;
- const auto& binlog_tablet_meta =
txn_info->attach_row_binlog.tablet->tablet_meta();
build_row_binlog =
- is_partial_update ||
binlog_tablet_meta->binlog_config().need_historical_value();
+ is_partial_update ||
+
txn_info->attach_row_binlog.tablet->binlog_config().need_historical_value();
}
// rewrite conflict only when partial update or need before
diff --git a/be/src/storage/tablet/base_tablet.h
b/be/src/storage/tablet/base_tablet.h
index 0bf620b7e15..ef5eb1e100b 100644
--- a/be/src/storage/tablet/base_tablet.h
+++ b/be/src/storage/tablet/base_tablet.h
@@ -99,6 +99,11 @@ public:
// Property encapsulated in TabletMeta
const TabletMetaSharedPtr& tablet_meta() const { return _tablet_meta; }
+ BinlogConfig binlog_config() const {
+ std::shared_lock rlock(_meta_lock);
+ return _tablet_meta->binlog_config();
+ }
+
int32_t max_version_config();
// FIXME(plat1ko): It is not appropriate to expose this lock
diff --git a/be/src/storage/tablet/tablet.cpp b/be/src/storage/tablet/tablet.cpp
index 9da3e99b512..9a549b84ac1 100644
--- a/be/src/storage/tablet/tablet.cpp
+++ b/be/src/storage/tablet/tablet.cpp
@@ -1019,8 +1019,7 @@ bool Tablet::can_do_compaction(size_t path_hash,
CompactionType compaction_type)
return false;
}
- if (compaction_type == CompactionType::CUMU_BINLOG_COMPACTION &&
- (!config::enable_feature_binlog || !is_row_binlog_tablet())) {
+ if (compaction_type == CompactionType::CUMU_BINLOG_COMPACTION &&
!is_row_binlog_tablet()) {
return false;
}
diff --git a/be/test/cloud/cloud_compaction_test.cpp
b/be/test/cloud/cloud_compaction_test.cpp
index 9f0ac135bc0..f0e4ad7a3a2 100644
--- a/be/test/cloud/cloud_compaction_test.cpp
+++ b/be/test/cloud/cloud_compaction_test.cpp
@@ -389,6 +389,56 @@ TEST_F(CloudCompactionTest,
generate_cloud_binlog_compaction_tasks_updates_only_
ASSERT_EQ(metrics->tablet_time_series_max_compaction_score->value(), 103);
}
+TEST_F(CloudCompactionTest, binlog_compaction_producer_ignores_ccr_switch) {
+ const bool old_enable_feature_binlog = config::enable_feature_binlog;
+ const bool old_disable_auto_compaction = config::disable_auto_compaction;
+ const int32_t old_binlog_compaction_task_num_per_disk =
+ config::binlog_compaction_task_num_per_disk;
+ Defer restore_config {[&]() {
+ config::enable_feature_binlog = old_enable_feature_binlog;
+ config::disable_auto_compaction = old_disable_auto_compaction;
+ config::binlog_compaction_task_num_per_disk =
old_binlog_compaction_task_num_per_disk;
+ }};
+
+ ASSERT_TRUE(ThreadPoolBuilder("BaseCompactionTaskThreadPoolTest")
+ .set_min_threads(1)
+ .set_max_threads(1)
+ .build(&_engine._base_compaction_thread_pool)
+ .ok());
+ ASSERT_TRUE(ThreadPoolBuilder("CumuCompactionTaskThreadPoolTest")
+ .set_min_threads(1)
+ .set_max_threads(1)
+ .build(&_engine._cumu_compaction_thread_pool)
+ .ok());
+ ASSERT_TRUE(ThreadPoolBuilder("BinlogCompactionTaskThreadPoolTest")
+ .set_min_threads(1)
+ .set_max_threads(1)
+ .build(&_engine._binlog_compaction_thread_pool)
+ .ok());
+
+ auto binlog_meta = std::make_shared<TabletMeta>(*_tablet_meta);
+ binlog_meta->_tablet_id = 11004;
+ binlog_meta->set_tablet_role(TabletRolePB::TABLET_ROLE_ROW_BINLOG);
+ auto binlog_tablet = std::make_shared<CloudTablet>(_engine, binlog_meta);
+
binlog_tablet->tablet_meta()->tablet_schema()->set_disable_auto_compaction(false);
+ binlog_tablet->_approximate_cumu_num_deltas = 7;
+ _engine.tablet_mgr().put_tablet_for_UT(binlog_tablet);
+
+ config::enable_feature_binlog = false;
+ config::disable_auto_compaction = false;
+ config::binlog_compaction_task_num_per_disk = 0;
+ auto* metric =
DorisMetrics::instance()->tablet_binlog_max_compaction_score;
+ metric->set_value(0);
+ _engine._stop_background_threads_latch.count_down();
+ _engine._binlog_compaction_tasks_producer_callback();
+ EXPECT_EQ(metric->value(), 7);
+
+ config::disable_auto_compaction = true;
+ metric->set_value(11);
+ _engine._binlog_compaction_tasks_producer_callback();
+ EXPECT_EQ(metric->value(), 11);
+}
+
TEST_F(CloudCompactionTest,
generate_cloud_compaction_tasks_clears_metrics_without_tablets) {
auto* metrics = DorisMetrics::instance();
metrics->tablet_cumulative_max_compaction_score->set_value(101);
diff --git a/be/test/cloud/cloud_tablet_test.cpp
b/be/test/cloud/cloud_tablet_test.cpp
index 3d3982286d5..90ef62825be 100644
--- a/be/test/cloud/cloud_tablet_test.cpp
+++ b/be/test/cloud/cloud_tablet_test.cpp
@@ -24,6 +24,8 @@
#include <chrono>
#include <cstdint>
+#include <future>
+#include <mutex>
#include "cloud/cloud_meta_mgr.h"
#include "cloud/cloud_storage_engine.h"
@@ -931,6 +933,8 @@ TEST_F(CloudTabletSyncMetaTest,
TestSyncMetaMultipleProperties) {
mock_tablet_meta->set_time_series_compaction_empty_rowsets_threshold(9);
mock_tablet_meta->set_time_series_compaction_level_threshold(7);
mock_tablet_meta->set_vertical_compaction_num_columns_per_group(13);
+ mock_tablet_meta->set_binlog_config(
+ BinlogConfig(true, 3600, 4096, 7, BinlogFormatPB::ROW, true));
// Mock get_tablet_meta to return tablet_meta with updated properties
sp->set_call_back("CloudMetaMgr::get_tablet_meta",
[mock_tablet_meta](auto&& args) {
@@ -954,11 +958,37 @@ TEST_F(CloudTabletSyncMetaTest,
TestSyncMetaMultipleProperties) {
EXPECT_EQ(_tablet->tablet_meta()->time_series_compaction_empty_rowsets_threshold(),
9);
EXPECT_EQ(_tablet->tablet_meta()->time_series_compaction_level_threshold(), 7);
EXPECT_EQ(_tablet->tablet_meta()->vertical_compaction_num_columns_per_group(),
13);
+ auto binlog_config = _tablet->binlog_config();
+ EXPECT_TRUE(binlog_config.is_enable());
+ EXPECT_EQ(binlog_config.ttl_seconds(), 3600);
+ EXPECT_EQ(binlog_config.max_bytes(), 4096);
+ EXPECT_EQ(binlog_config.max_history_nums(), 7);
+ EXPECT_EQ(binlog_config.binlog_format(), BinlogFormatPB::ROW);
+ EXPECT_TRUE(binlog_config.need_historical_value());
sp->disable_processing();
sp->clear_all_call_backs();
}
+TEST_F(CloudTabletSyncMetaTest, TestBinlogConfigReadUsesMetaLock) {
+ std::unique_lock meta_lock(_tablet->get_header_lock());
+ std::promise<void> reader_started;
+ auto reader_started_future = reader_started.get_future();
+ auto read_future = std::async(std::launch::async, [&]() {
+ reader_started.set_value();
+ return _tablet->binlog_config();
+ });
+
+ auto reader_started_status = reader_started_future.wait_for(seconds(5));
+ auto read_blocked_status = read_future.wait_for(milliseconds(100));
+ meta_lock.unlock();
+
+ EXPECT_EQ(reader_started_status, std::future_status::ready);
+ EXPECT_EQ(read_blocked_status, std::future_status::timeout);
+ ASSERT_EQ(read_future.wait_for(seconds(5)), std::future_status::ready);
+ EXPECT_EQ(read_future.get(), _tablet->binlog_config());
+}
+
TEST_F(CloudTabletSyncMetaTest, TestSyncMetaSyncsTtlWithoutChangingInMemory) {
EXPECT_EQ(0, _tablet->tablet_meta()->ttl_seconds());
EXPECT_FALSE(_tablet->tablet_meta()->tablet_schema()->is_in_memory());
diff --git a/be/test/storage/compaction/compaction_task_test.cpp
b/be/test/storage/compaction/compaction_task_test.cpp
index 85ae15a7357..3f2c1d29c99 100644
--- a/be/test/storage/compaction/compaction_task_test.cpp
+++ b/be/test/storage/compaction/compaction_task_test.cpp
@@ -130,6 +130,46 @@ TEST_F(CompactionTaskTest, TestSubmitCompactionTask) {
EXPECT_EQ(executing_task_num, 2);
}
+TEST_F(CompactionTaskTest, BinlogCompactionProducerIgnoresCcrSwitch) {
+ const bool old_enable_feature_binlog = config::enable_feature_binlog;
+ const bool old_disable_auto_compaction = config::disable_auto_compaction;
+ const bool old_enable_compaction_pause_on_high_memory =
+ config::enable_compaction_pause_on_high_memory;
+ Defer restore_config {[&]() {
+ config::enable_feature_binlog = old_enable_feature_binlog;
+ config::disable_auto_compaction = old_disable_auto_compaction;
+ config::enable_compaction_pause_on_high_memory =
old_enable_compaction_pause_on_high_memory;
+ }};
+
+ auto* sp = SyncPoint::get_instance();
+ sp->enable_processing();
+ int generation_count = 0;
+
sp->set_call_back("StorageEngine::_adjust_compaction_thread_num.return_void",
+ [](auto&& args) { *try_any_cast<bool*>(args.back()) =
true; });
+ sp->set_call_back("olap_server::_generate_compaction_tasks.return_empty",
+ [&generation_count](auto&& values) {
+ ++generation_count;
+ auto* ret =
+
try_any_cast_ret<std::vector<TabletCompactionContext>>(values);
+ ret->second = true;
+ });
+ Defer clear_sync_points {[&]() {
+ sp->clear_all_call_backs();
+ sp->disable_processing();
+ }};
+
+ config::enable_feature_binlog = false;
+ config::disable_auto_compaction = false;
+ config::enable_compaction_pause_on_high_memory = false;
+ _storage_engine->_stop_background_threads_latch.count_down();
+ _storage_engine->_binlog_compaction_tasks_producer_callback();
+ EXPECT_EQ(generation_count, 1);
+
+ config::disable_auto_compaction = true;
+ _storage_engine->_binlog_compaction_tasks_producer_callback();
+ EXPECT_EQ(generation_count, 1);
+}
+
TEST_F(CompactionTaskTest, TestAutoSetCompactionIncreaseTaskNum) {
auto st = ThreadPoolBuilder("BaseCompactionTaskThreadPool")
.set_min_threads(2)
diff --git a/be/test/storage/tablet/tablet_test.cpp
b/be/test/storage/tablet/tablet_test.cpp
index ea4719e31c9..21c6d91afba 100644
--- a/be/test/storage/tablet/tablet_test.cpp
+++ b/be/test/storage/tablet/tablet_test.cpp
@@ -26,6 +26,7 @@
#include <memory>
+#include "common/config.h"
#include "gtest/gtest_pred_impl.h"
#include "io/fs/local_file_system.h"
#include "json2pb/json_to_pb.h"
@@ -39,6 +40,7 @@
#include "storage/tablet/tablet_meta.h"
#include "storage/utils.h"
#include "testutil/mock_rowset.h"
+#include "util/defer_op.h"
#include "util/time.h"
#include "util/uid_util.h"
@@ -269,6 +271,30 @@ TEST_F(TestTablet, delete_expired_stale_rowset) {
_tablet.reset();
}
+TEST_F(TestTablet, RowBinlogCompactionDoesNotDependOnCcrSwitch) {
+ const bool old_enable_feature_binlog = config::enable_feature_binlog;
+ Defer restore_config {[&]() { config::enable_feature_binlog =
old_enable_feature_binlog; }};
+ config::enable_feature_binlog = false;
+
+ ASSERT_TRUE(_data_dir->init().ok());
+ _tablet_meta->set_tablet_role(TabletRolePB::TABLET_ROLE_ROW_BINLOG);
+ TabletSharedPtr row_binlog_tablet(
+ new Tablet(*k_engine, _tablet_meta, _data_dir.get(),
CUMULATIVE_SIZE_BASED_POLICY));
+ ASSERT_TRUE(row_binlog_tablet->init().ok());
+ EXPECT_TRUE(row_binlog_tablet->can_do_compaction(_data_dir->path_hash(),
+
CompactionType::CUMU_BINLOG_COMPACTION));
+ EXPECT_FALSE(row_binlog_tablet->can_do_compaction(_data_dir->path_hash(),
+
CompactionType::CUMULATIVE_COMPACTION));
+
+ auto data_tablet_meta = new_tablet_meta(TTabletSchema());
+ data_tablet_meta->set_tablet_role(TabletRolePB::TABLET_ROLE_DATA);
+ TabletSharedPtr data_tablet(
+ new Tablet(*k_engine, data_tablet_meta, _data_dir.get(),
CUMULATIVE_SIZE_BASED_POLICY));
+ ASSERT_TRUE(data_tablet->init().ok());
+ EXPECT_FALSE(data_tablet->can_do_compaction(_data_dir->path_hash(),
+
CompactionType::CUMU_BINLOG_COMPACTION));
+}
+
TEST_F(TestTablet, pad_rowset) {
std::vector<RowsetMetaSharedPtr> rs_metas;
auto ptr1 = std::make_shared<RowsetMeta>();
diff --git a/cloud/src/meta-service/meta_service.cpp
b/cloud/src/meta-service/meta_service.cpp
index c6aa71e38c8..2d6bd800223 100644
--- a/cloud/src/meta-service/meta_service.cpp
+++ b/cloud/src/meta-service/meta_service.cpp
@@ -1204,6 +1204,8 @@ void
MetaServiceImpl::update_tablet(::google::protobuf::RpcController* controlle
tablet_meta.set_is_persistent(tablet_meta_info.is_persistent());
} else if (tablet_meta_info.has_ttl_seconds()) {
tablet_meta.set_ttl_seconds(tablet_meta_info.ttl_seconds());
+ } else if (tablet_meta_info.has_binlog_config()) {
+
tablet_meta.mutable_binlog_config()->CopyFrom(tablet_meta_info.binlog_config());
} else if (tablet_meta_info.has_compaction_policy()) {
tablet_meta.set_compaction_policy(tablet_meta_info.compaction_policy());
} else if
(tablet_meta_info.has_time_series_compaction_goal_size_mbytes()) {
diff --git a/cloud/test/meta_service_test.cpp b/cloud/test/meta_service_test.cpp
index 25ac956415c..15f5483cf18 100644
--- a/cloud/test/meta_service_test.cpp
+++ b/cloud/test/meta_service_test.cpp
@@ -5122,6 +5122,40 @@ TEST(MetaServiceTest, UpdateTablet) {
ASSERT_EQ(resp.status().code(), MetaServiceCode::OK);
}
get_and_check_tablet_meta(tablet_id1, 300, true, true);
+ {
+ brpc::Controller cntl;
+ UpdateTabletRequest req;
+ UpdateTabletResponse resp;
+ req.set_cloud_unique_id(cloud_unique_id);
+ TabletMetaInfoPB* tablet_meta_info = req.add_tablet_meta_infos();
+ tablet_meta_info->set_tablet_id(tablet_id1);
+ auto* binlog_config = tablet_meta_info->mutable_binlog_config();
+ binlog_config->set_enable(true);
+ binlog_config->set_ttl_seconds(3600);
+ binlog_config->set_max_bytes(4096);
+ binlog_config->set_max_history_nums(7);
+ binlog_config->set_binlog_format(BinlogFormatPB::ROW);
+ binlog_config->set_need_historical_value(true);
+ meta_service->update_tablet(&cntl, &req, &resp, nullptr);
+ ASSERT_EQ(resp.status().code(), MetaServiceCode::OK);
+ }
+ {
+ brpc::Controller cntl;
+ GetTabletRequest req;
+ req.set_cloud_unique_id(cloud_unique_id);
+ req.set_tablet_id(tablet_id1);
+ GetTabletResponse resp;
+ meta_service->get_tablet(&cntl, &req, &resp, nullptr);
+ ASSERT_EQ(resp.status().code(), MetaServiceCode::OK);
+ ASSERT_TRUE(resp.tablet_meta().has_binlog_config());
+ const auto& binlog_config = resp.tablet_meta().binlog_config();
+ EXPECT_TRUE(binlog_config.enable());
+ EXPECT_EQ(binlog_config.ttl_seconds(), 3600);
+ EXPECT_EQ(binlog_config.max_bytes(), 4096);
+ EXPECT_EQ(binlog_config.max_history_nums(), 7);
+ EXPECT_EQ(binlog_config.binlog_format(), BinlogFormatPB::ROW);
+ EXPECT_TRUE(binlog_config.need_historical_value());
+ }
}
TEST(MetaServiceTest, GetTabletStatsTest) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/TableStreamManager.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/TableStreamManager.java
index f768f5afeb9..35275748397 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/TableStreamManager.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/TableStreamManager.java
@@ -63,6 +63,7 @@ import java.util.concurrent.locks.LockSupport;
public class TableStreamManager extends MasterDaemon implements Writable,
GsonPostProcessable {
private static final Logger LOG =
LogManager.getLogger(TableStreamManager.class);
+ private static final String BASE_TABLE_NOT_FOUND_STALE_REASON = "Base
table does not exist";
@SerializedName(value = "dbStreamMap")
private Map<Long, Set<Long>> dbStreamMap;
protected MonitoredReentrantReadWriteLock rwLock;
@@ -356,7 +357,8 @@ public class TableStreamManager extends MasterDaemon
implements Writable, GsonPo
// STREAM_COMMENT
trow.addToColumnValue(new
TCell().setStringVal(stream.getComment()));
TableIf baseTable = stream.getBaseTableNullable();
- if (baseTable == null) {
+ boolean baseTableExists = baseTable != null;
+ if (!baseTableExists) {
// BASE_TABLE_NAME
trow.addToColumnValue(new
TCell().setStringVal("N/A"));
// BASE_TABLE_DB
@@ -377,11 +379,13 @@ public class TableStreamManager extends MasterDaemon
implements Writable, GsonPo
trow.addToColumnValue(new
TCell().setStringVal(baseTable.getType().name()));
}
// ENABLED
- trow.addToColumnValue(new
TCell().setBoolVal(!stream.isDisabled()));
+ trow.addToColumnValue(
+ new TCell().setBoolVal(baseTableExists &&
!stream.isDisabled()));
// IS_STALE
- trow.addToColumnValue(new
TCell().setBoolVal(stream.isStale()));
+ trow.addToColumnValue(new
TCell().setBoolVal(!baseTableExists || stream.isStale()));
// STALE_REASON
- trow.addToColumnValue(new
TCell().setStringVal(stream.getStaleReason()));
+ trow.addToColumnValue(new
TCell().setStringVal(baseTableExists
+ ? stream.getStaleReason() :
BASE_TABLE_NOT_FOUND_STALE_REASON));
dataBatch.add(trow);
} finally {
stream.readUnlock();
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/alter/CloudSchemaChangeHandler.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/alter/CloudSchemaChangeHandler.java
index 41c9827cb9a..ed6008c8391 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/cloud/alter/CloudSchemaChangeHandler.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/cloud/alter/CloudSchemaChangeHandler.java
@@ -18,6 +18,7 @@
package org.apache.doris.cloud.alter;
import org.apache.doris.alter.SchemaChangeHandler;
+import org.apache.doris.catalog.BinlogConfig;
import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.KeysType;
@@ -426,6 +427,18 @@ public class CloudSchemaChangeHandler extends
SchemaChangeHandler {
DynamicPartitionUtil.registerOrRemoveDynamicPartitionTable(db.getId(),
olapTable, false);
}
+ @Override
+ public void updatePartitionProperties(Database db, String tableName,
String partitionName,
+ long storagePolicyId, int isInMemory, BinlogConfig binlogConfig,
String compactionPolicy,
+ Map<String, Long> timeSeriesCompactionConfig, int
skipWriteIndexOnLoad,
+ int disableAutoCompaction, int
verticalCompactionNumColumnsPerGroup) throws UserException {
+ Preconditions.checkNotNull(binlogConfig);
+ UpdatePartitionMetaParam param = new UpdatePartitionMetaParam();
+ param.binlogConfig = binlogConfig;
+ param.type = UpdatePartitionMetaParam.TabletMetaType.BINLOG_CONFIG;
+ updateCloudPartitionMeta(db, tableName, partitionName, param);
+ }
+
private static class UpdatePartitionMetaParam {
public enum TabletMetaType {
INMEMORY,
@@ -443,6 +456,7 @@ public class CloudSchemaChangeHandler extends
SchemaChangeHandler {
DISABLE_AUTO_COMPACTION,
ENABLE_MOW_LIGHT_DELETE,
VERTICAL_COMPACTION_NUM_COLUMNS_PER_GROUP,
+ BINLOG_CONFIG,
}
TabletMetaType type;
@@ -461,6 +475,7 @@ public class CloudSchemaChangeHandler extends
SchemaChangeHandler {
boolean disableAutoCompaction = false;
boolean enableMowLightDelete = false;
int verticalCompactionNumColumnsPerGroup = 5;
+ BinlogConfig binlogConfig;
}
public void updateCloudPartitionMeta(Database db,
@@ -476,7 +491,9 @@ public class CloudSchemaChangeHandler extends
SchemaChangeHandler {
throw new DdlException(
"Partition[" + partitionName + "] does not exist in
table[" + olapTable.getName() + "]");
}
- for (MaterializedIndex index :
partition.getMaterializedIndices(IndexExtState.VISIBLE, true)) {
+ boolean includeRowBinlog = param.type !=
UpdatePartitionMetaParam.TabletMetaType.COMPACTION_POLICY;
+ for (MaterializedIndex index
+ : partition.getMaterializedIndices(IndexExtState.VISIBLE,
includeRowBinlog)) {
for (Tablet tablet : index.getTablets()) {
tabletIds.add(tablet.getId());
}
@@ -548,6 +565,9 @@ public class CloudSchemaChangeHandler extends
SchemaChangeHandler {
infoBuilder.setVerticalCompactionNumColumnsPerGroup(
param.verticalCompactionNumColumnsPerGroup);
break;
+ case BINLOG_CONFIG:
+
infoBuilder.setBinlogConfig(param.binlogConfig.toProtobuf());
+ break;
default:
throw new UserException("Unknown TabletMetaType");
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/DeleteHandler.java
b/fe/fe-core/src/main/java/org/apache/doris/load/DeleteHandler.java
index 8f2aee7f9af..aed66e1508d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/DeleteHandler.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/DeleteHandler.java
@@ -20,7 +20,6 @@ package org.apache.doris.load;
import org.apache.doris.analysis.Predicate;
import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.Env;
-import org.apache.doris.catalog.KeysType;
import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.Partition;
import org.apache.doris.common.AnalysisException;
@@ -112,10 +111,10 @@ public class DeleteHandler implements Writable {
try {
targetTbl.readLock();
try {
- if (targetTbl.needRowBinlog() && targetTbl.getKeysType() ==
KeysType.DUP_KEYS) {
+ if (targetTbl.needRowBinlog()) {
throw new AnalysisException(
- "DELETE with predicates is not supported on
DUPLICATE KEY tables when binlog<row>"
- + "is enabled. Please disable binlog<row>
for this table or avoid DELETE.");
+ "DELETE with predicates is not supported when
binlog<row> is enabled because it cannot "
+ + "produce row-level DELETE events.");
}
if (targetTbl.getState() != OlapTable.OlapTableState.NORMAL) {
// table under alter operation can also do delete.
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java
index 60479e23a8b..6b5a8628c6b 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java
@@ -59,6 +59,7 @@ import org.apache.doris.nereids.StatementContext.TableFrom;
import org.apache.doris.nereids.analyzer.Unbound;
import org.apache.doris.nereids.analyzer.UnboundRelation;
import org.apache.doris.nereids.analyzer.UnboundResultSink;
+import org.apache.doris.nereids.analyzer.UnboundSlot;
import org.apache.doris.nereids.exceptions.AnalysisException;
import org.apache.doris.nereids.exceptions.ParseException;
import org.apache.doris.nereids.hint.LeadingHint;
@@ -98,6 +99,7 @@ import org.apache.doris.nereids.trees.plans.algebra.Relation;
import org.apache.doris.nereids.trees.plans.algebra.SetOperation.Qualifier;
import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
import org.apache.doris.nereids.trees.plans.logical.LogicalCTEConsumer;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCheckPolicy;
import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan;
import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
import org.apache.doris.nereids.trees.plans.logical.LogicalOdbcScan;
@@ -587,22 +589,27 @@ public class BindRelation extends OneAnalysisRuleFactory {
}
/**
- * mow time-travel: A|t1 = base(survived rows, tso<=t1) UNION ALL
binlog(before-image of
- * UPDATE_BEFORE/DELETE since t1). The binlog right branch reuses the
@incr (MIN_DELTA) machinery;
+ * mow time-travel: A|t1 = base(survived rows, tso < targetTso) UNION
ALL binlog(before-image of
+ * UPDATE_BEFORE/DELETE from targetTso). The binlog right branch reuses
the @incr (MIN_DELTA) machinery;
* BE splits each change into rows where UPDATE_BEFORE/DELETE rows already
carry the before value.
*/
private LogicalPlan buildMowTimeTravelUnion(LogicalOlapScan baseScan,
OlapTable olapTable,
long targetTso, UnboundRelation unboundRelation, List<String>
qualifier,
List<Long> partIds, List<Long> tabletIds, CascadesContext
cascadesContext) {
- // union baseline = base visible columns (key + value); hidden cols
are filtered out.
- List<Slot> visibleOutput = baseScan.getOutput().stream()
+ // Use unbound visible columns so each branch projection binds after
its policy is expanded.
+ // Otherwise the projections keep the scan's raw slots and can bypass
data masking.
+ // Keep the original qualifier in the UnboundSlot name parts so
projectFromUnboundSlots can
+ // preserve it on the projection alias.
+ List<UnboundSlot> visibleOutput = baseScan.getOutput().stream()
.filter(slot -> !(slot instanceof SlotReference)
|| ((SlotReference) slot).isVisible())
+ .map(slot -> new
UnboundSlot(Utils.qualifiedNameParts(slot.getQualifier(), slot.getName())))
.collect(Collectors.toList());
// left: base survived rows at t1 = delete_sign=0 AND commit_tso <
targetTso, projected to visible.
LogicalPlan left = checkAndAddDeleteSignFilter(baseScan,
ConnectContext.get(), olapTable, true);
- left = projectFromOriginSlots(addCommitTsoFilter(left, targetTso,
olapTable), visibleOutput);
+ left = addCommitTsoFilter(left, targetTso, olapTable);
+ left = projectFromUnboundSlots(new LogicalCheckPolicy<>(left),
visibleOutput);
// right: binlog MIN_DELTA over tso >= targetTso, keep
UPDATE_BEFORE/DELETE rows (before image),
// projected to the same visible schema. BE splits each change so
UPDATE_BEFORE/DELETE rows
@@ -625,9 +632,9 @@ public class BindRelation extends OneAnalysisRuleFactory {
new TableScanParams(TableScanParams.INCREMENTAL_READ,
incrParams, Lists.newArrayList()));
LogicalPlan right = checkAndAddChangeScanFilter(binlogScan,
StreamScanType.MIN_DELTA, true);
- right = projectFromOriginSlots(right, visibleOutput);
+ right = projectFromUnboundSlots(new LogicalCheckPolicy<>(right),
visibleOutput);
- // both children are bound; BindExpression aligns by position and
fills the union output.
+ // BindExpression binds both branch projections, aligns them by
position, and fills the union output.
// buildNewOutputs() rebuilds the union output slots with empty
qualifiers, so wrap the union
// in a subquery alias to restore the original catalog.db.table
qualifier. Use the scan's
// fully-qualified name (catalog.db.table) rather than the table-less
qualifier, otherwise
@@ -1057,7 +1064,6 @@ public class BindRelation extends OneAnalysisRuleFactory {
*/
public static LogicalPlan checkAndAddChangeScanFilter(LogicalOlapScan scan,
StreamScanType
scanType, boolean beforeImageOnly) {
- LogicalPlan plan = scan;
Slot opSlot = null;
for (Slot slot : scan.getOutput()) {
if (slot.getName().equals(Column.BINLOG_OPERATION_COL)) {
@@ -1068,26 +1074,22 @@ public class BindRelation extends
OneAnalysisRuleFactory {
if (scanType.equals(StreamScanType.APPEND_ONLY)) {
Preconditions.checkArgument(opSlot != null, "opSlot is null");
return new LogicalFilter<>(ImmutableSet.of(new EqualTo(opSlot,
- new BigIntLiteral(BinlogUtils.ROW_BINLOG_APPEND))), plan);
+ new BigIntLiteral(BinlogUtils.ROW_BINLOG_APPEND))), scan);
} else if (beforeImageOnly) {
return new LogicalFilter<>(ImmutableSet.of(new InPredicate(opSlot,
ImmutableList.of(
new BigIntLiteral(BinlogUtils.ROW_BINLOG_DELETE),
- new
BigIntLiteral(BinlogUtils.ROW_BINLOG_UPDATE_BEFORE)))), plan);
+ new
BigIntLiteral(BinlogUtils.ROW_BINLOG_UPDATE_BEFORE)))), scan);
}
- return plan;
+ return scan;
}
- private LogicalPlan projectFromOriginSlots(LogicalPlan plan, List<Slot>
wantedSlots) {
- Map<String, Slot> childSlotByName = new
HashMap<>(plan.getOutput().size());
- for (Slot slot : plan.getOutput()) {
- childSlotByName.put(slot.getName(), slot);
- }
+ private LogicalPlan projectFromUnboundSlots(LogicalPlan plan,
List<UnboundSlot> wantedSlots) {
List<NamedExpression> project = new ArrayList<>(wantedSlots.size());
- for (Slot wanted : wantedSlots) {
- Slot match = childSlotByName.get(wanted.getName());
- Preconditions.checkArgument(match != null,
- "column %s not found in child output", wanted.getName());
- project.add(new Alias(match, wanted.getName()));
+ for (UnboundSlot wanted : wantedSlots) {
+ List<String> nameParts = wanted.getNameParts();
+ int nameIndex = nameParts.size() - 1;
+ project.add(new Alias(wanted, nameParts.get(nameIndex),
+ nameParts.subList(0, nameIndex)));
}
return new LogicalProject<>(project, plan);
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/CheckPolicy.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/CheckPolicy.java
index 0fb1c9a32e0..2577c2770b5 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/CheckPolicy.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/CheckPolicy.java
@@ -61,20 +61,15 @@ public class CheckPolicy implements AnalysisRuleFactory {
RuleType.CHECK_ROW_POLICY.build(
logicalCheckPolicy(any().when(child -> !(child
instanceof UnboundRelation))).thenApply(ctx -> {
LogicalCheckPolicy<Plan> checkPolicy = ctx.root;
- LogicalFilter<Plan> upperFilter = null;
Plan upAgg = null;
Plan child = checkPolicy.child();
- // Because the unique table will automatically
include a filter condition
- if ((child instanceof LogicalFilter)) {
- upperFilter = (LogicalFilter) child;
- if (child.child(0) instanceof LogicalRelation)
{
- child = child.child(0);
- } else if (child.child(0) instanceof
LogicalAggregate
- && child.child(0).child(0) instanceof
LogicalRelation) {
- upAgg = child.child(0);
- child = child.child(0).child(0);
- }
+ Set<Expression> upperFilterConjuncts = new
LinkedHashSet<>();
+ // Unique tables and time-travel reconstruction
can add a chain of filters around
+ // the relation. Keep all of them below a policy
mask when the policy is expanded.
+ while (child instanceof LogicalFilter) {
+
upperFilterConjuncts.addAll(((LogicalFilter<?>) child).getConjuncts());
+ child = child.child(0);
}
if ((child instanceof LogicalAggregate) &&
child.child(0) instanceof LogicalRelation) {
upAgg = child;
@@ -105,9 +100,7 @@ public class CheckPolicy implements AnalysisRuleFactory {
relatedPolicy.rowPolicyFilter.ifPresent(expression
-> combineFilter.addAll(
ExpressionUtils.extractConjunctionToSet(expression)));
Plan result = upAgg != null ?
upAgg.withChildren(child) : child;
- if (upperFilter != null) {
-
combineFilter.addAll(upperFilter.getConjuncts());
- }
+ combineFilter.addAll(upperFilterConjuncts);
if (!combineFilter.isEmpty()) {
result = new LogicalFilter<>(combineFilter,
result);
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DeleteFromCommand.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DeleteFromCommand.java
index 8b426283e6b..bda236a15bc 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DeleteFromCommand.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DeleteFromCommand.java
@@ -210,9 +210,10 @@ public class DeleteFromCommand extends Command implements
ForwardWithSync, Expla
}
}
- // if table's enable_mow_light_delete is false, use
`DeleteFromUsingCommand`
+ // Row binlog needs row-bearing deletes to emit DELETE events.
Predicate deletes only write
+ // delete predicates, so use `DeleteFromUsingCommand` even when MOW
light delete is enabled.
if (olapTable.getKeysType() == KeysType.UNIQUE_KEYS &&
olapTable.getEnableUniqueKeyMergeOnWrite()
- && !olapTable.getEnableMowLightDelete()) {
+ && (!olapTable.getEnableMowLightDelete() ||
olapTable.needRowBinlog())) {
new DeleteFromUsingCommand(nameParts, tableAlias, isTempPart,
partitions, logicalQuery,
Optional.empty(), false).run(ctx, executor);
return;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalCheckPolicy.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalCheckPolicy.java
index 6d204304cb1..920aca603f0 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalCheckPolicy.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalCheckPolicy.java
@@ -19,6 +19,7 @@ package org.apache.doris.nereids.trees.plans.logical;
import org.apache.doris.analysis.UserIdentity;
import org.apache.doris.catalog.DatabaseIf;
+import org.apache.doris.catalog.OlapTableWrapper;
import org.apache.doris.catalog.TableIf;
import org.apache.doris.datasource.CatalogIf;
import org.apache.doris.mysql.privilege.AccessControllerManager;
@@ -146,10 +147,11 @@ public class LogicalCheckPolicy<CHILD_TYPE extends Plan>
extends LogicalUnary<CH
if (!(logicalPlan instanceof CatalogRelation || logicalPlan instanceof
LogicalView)) {
return RelatedPolicy.NO_POLICY;
}
+ TableIf table = getPolicyTable(logicalPlan);
Optional<Map<TableIf, Set<Expression>>> mvRefreshPredicates =
cascadesContext.getStatementContext()
.getMvRefreshPredicates();
if (mvRefreshPredicates.isPresent()) {
- return findPolicyByMvRefresh(mvRefreshPredicates.get(),
logicalPlan);
+ return findPolicyByMvRefresh(mvRefreshPredicates.get(), table);
}
ConnectContext connectContext = cascadesContext.getConnectContext();
AccessControllerManager accessManager =
connectContext.getEnv().getAccessManager();
@@ -158,8 +160,6 @@ public class LogicalCheckPolicy<CHILD_TYPE extends Plan>
extends LogicalUnary<CH
return RelatedPolicy.NO_POLICY;
}
- TableIf table = logicalPlan instanceof CatalogRelation ?
((CatalogRelation) logicalPlan).getTable()
- : ((LogicalView<?>) logicalPlan).getView();
DatabaseIf database = table.getDatabase();
if (database == null) {
return RelatedPolicy.NO_POLICY;
@@ -213,10 +213,19 @@ public class LogicalCheckPolicy<CHILD_TYPE extends Plan>
extends LogicalUnary<CH
);
}
- private RelatedPolicy findPolicyByMvRefresh(Map<TableIf, Set<Expression>>
mvRefreshPredicates,
- LogicalPlan logicalPlan) {
+ private static TableIf getPolicyTable(LogicalPlan logicalPlan) {
TableIf table = logicalPlan instanceof CatalogRelation ?
((CatalogRelation) logicalPlan).getTable()
: ((LogicalView<?>) logicalPlan).getView();
+ // A wrapper changes how a scan reads data, not which table owns its
authorization policy. Unwrap
+ // recursively so future wrappers and stacked wrappers keep using the
original table identity.
+ while (table instanceof OlapTableWrapper) {
+ table = ((OlapTableWrapper) table).getOriginTable();
+ }
+ return table;
+ }
+
+ private RelatedPolicy findPolicyByMvRefresh(Map<TableIf, Set<Expression>>
mvRefreshPredicates,
+ TableIf table) {
if (mvRefreshPredicates.containsKey(table)) {
return new
RelatedPolicy(Optional.of(ExpressionUtils.or(mvRefreshPredicates.get(table))),
Optional.empty());
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/catalog/DropTableStreamTest.java
b/fe/fe-core/src/test/java/org/apache/doris/catalog/DropTableStreamTest.java
index 83a43edfb00..5e223251ecc 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/catalog/DropTableStreamTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/catalog/DropTableStreamTest.java
@@ -17,6 +17,7 @@
package org.apache.doris.catalog;
+import org.apache.doris.catalog.stream.OlapTableStream;
import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.ExceptionChecker;
@@ -25,11 +26,15 @@ import org.apache.doris.nereids.parser.NereidsParser;
import org.apache.doris.nereids.trees.plans.commands.DropStreamCommand;
import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
import org.apache.doris.qe.StmtExecutor;
+import org.apache.doris.thrift.TRow;
import org.apache.doris.utframe.TestWithFeService;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.util.ArrayList;
+import java.util.List;
+
public class DropTableStreamTest extends TestWithFeService {
@Override
@@ -67,6 +72,33 @@ public class DropTableStreamTest extends TestWithFeService {
}
}
+ private void createBaseTableAndStream(String tableName, String streamName)
throws Exception {
+ createTable("create table test_stream." + tableName + " (k1 int, k2
int) "
+ + "unique key(k1) distributed by hash(k1) buckets 1 "
+ + "properties('replication_num' = '1', 'binlog.enable' =
'true', 'binlog.format' = 'ROW', "
+ + "'binlog.need_historical_value' = 'true')");
+ createTable("create stream test_stream." + streamName + " on table
test_stream." + tableName
+ + " properties('show_initial_rows' = 'true')");
+ }
+
+ private TRow getStreamMetadataRow(String streamName) {
+ List<TRow> rows = new ArrayList<>();
+
Env.getCurrentEnv().getTableStreamManager().fillTableStreamValuesMetadataResult(rows);
+ return rows.stream()
+ .filter(row ->
streamName.equals(row.getColumnValue().get(1).getStringVal()))
+ .findFirst()
+ .orElseThrow(AssertionError::new);
+ }
+
+ private void assertStreamMetadataState(String streamName, String
baseTableType,
+ boolean enabled, boolean stale, String staleReason) {
+ TRow row = getStreamMetadataRow(streamName);
+ Assertions.assertEquals(baseTableType,
row.getColumnValue().get(9).getStringVal());
+ Assertions.assertEquals(enabled,
row.getColumnValue().get(10).isBoolVal());
+ Assertions.assertEquals(stale,
row.getColumnValue().get(11).isBoolVal());
+ Assertions.assertEquals(staleReason,
row.getColumnValue().get(12).getStringVal());
+ }
+
@Test
public void testNormalDropStream() throws Exception {
// test drop
@@ -104,6 +136,43 @@ public class DropTableStreamTest extends TestWithFeService
{
}
}
+ @Test
+ public void testStreamMetadataFollowsRecoverableBaseTable() throws
Exception {
+ createBaseTableAndStream("tbl_recover", "s_recover");
+ Database db =
Env.getCurrentInternalCatalog().getDbOrMetaException("test_stream");
+ long baseTableId = db.getTableOrMetaException("tbl_recover").getId();
+
+ assertStreamMetadataState("s_recover", "OLAP", true, false, "N/A");
+
+ dropTableWithSql("drop table test_stream.tbl_recover");
+ assertStreamMetadataState("s_recover", "N/A", false, true, "Base table
does not exist");
+
+ recoverTable("recover table test_stream.tbl_recover");
+ Assertions.assertEquals(baseTableId,
db.getTableOrMetaException("tbl_recover").getId());
+ assertStreamMetadataState("s_recover", "OLAP", true, false, "N/A");
+ }
+
+ @Test
+ public void testStreamMetadataRejectsSameNameBaseTable() throws Exception {
+ createBaseTableAndStream("tbl_force", "s_force");
+ Database db =
Env.getCurrentInternalCatalog().getDbOrMetaException("test_stream");
+ long oldBaseTableId = db.getTableOrMetaException("tbl_force").getId();
+ OlapTableStream stream = (OlapTableStream)
db.getTableOrMetaException("s_force");
+
+ dropTableWithSql("drop table test_stream.tbl_force force");
+ Assertions.assertNull(stream.getBaseTableNullable());
+ assertStreamMetadataState("s_force", "N/A", false, true, "Base table
does not exist");
+
+ createTable("create table test_stream.tbl_force (k1 int, k2 int) "
+ + "unique key(k1) distributed by hash(k1) buckets 1 "
+ + "properties('replication_num' = '1', 'binlog.enable' =
'true', 'binlog.format' = 'ROW', "
+ + "'binlog.need_historical_value' = 'true')");
+
+ Assertions.assertNotEquals(oldBaseTableId,
db.getTableOrMetaException("tbl_force").getId());
+ Assertions.assertNull(stream.getBaseTableNullable());
+ assertStreamMetadataState("s_force", "N/A", false, true, "Base table
does not exist");
+ }
+
@Override
protected void runAfterAll() throws Exception {
dropDatabase("test_stream");
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/cloud/alter/CloudSchemaChangeHandlerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/cloud/alter/CloudSchemaChangeHandlerTest.java
index 885e39052fb..c5037746aec 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/cloud/alter/CloudSchemaChangeHandlerTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/cloud/alter/CloudSchemaChangeHandlerTest.java
@@ -17,6 +17,7 @@
package org.apache.doris.cloud.alter;
+import org.apache.doris.catalog.BinlogConfig;
import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.KeysType;
@@ -155,6 +156,104 @@ public class CloudSchemaChangeHandlerTest {
Assertions.assertEquals(Arrays.asList(103L),
syncRequests.get(3).getTabletIdsList());
}
+ @Test
+ public void testUpdateBinlogConfigUsesCloudTabletMeta() throws Exception {
+ CloudSchemaChangeHandler handler = new CloudSchemaChangeHandler();
+ Database db = createMockDatabaseWithThreeTablets();
+ BinlogConfig binlogConfig = new BinlogConfig(true, 3600L, 4096L, 7L,
+ BinlogConfig.BinlogFormat.ROW, true);
+
+ MetaServiceProxy metaServiceProxy =
Mockito.mock(MetaServiceProxy.class);
+
Mockito.when(metaServiceProxy.updateTablet(Mockito.any())).thenReturn(okUpdateTabletResponse());
+ Config.enable_debug_points = true;
+
DebugPointUtil.addDebugPoint("CloudSchemaChangeHandler.notifyBackendsToSyncTabletMeta.skip");
+
+ try (MockedStatic<MetaServiceProxy> metaProxyMock =
Mockito.mockStatic(MetaServiceProxy.class)) {
+
metaProxyMock.when(MetaServiceProxy::getInstance).thenReturn(metaServiceProxy);
+ handler.updatePartitionProperties(db, "tbl", "p1", -1, -1,
binlogConfig,
+ null, null, -1, -1, -1);
+ }
+
+ ArgumentCaptor<Cloud.UpdateTabletRequest> updateCaptor =
+ ArgumentCaptor.forClass(Cloud.UpdateTabletRequest.class);
+ Mockito.verify(metaServiceProxy,
Mockito.times(2)).updateTablet(updateCaptor.capture());
+ List<Cloud.TabletMetaInfoPB> tabletMetaInfos =
updateCaptor.getAllValues().stream()
+ .flatMap(request -> request.getTabletMetaInfosList().stream())
+ .collect(Collectors.toList());
+ Assertions.assertEquals(Arrays.asList(101L, 102L, 103L),
tabletMetaInfos.stream()
+
.map(Cloud.TabletMetaInfoPB::getTabletId).collect(Collectors.toList()));
+ for (Cloud.TabletMetaInfoPB tabletMetaInfo : tabletMetaInfos) {
+ Assertions.assertTrue(tabletMetaInfo.hasBinlogConfig());
+
Assertions.assertTrue(tabletMetaInfo.getBinlogConfig().getEnable());
+ Assertions.assertEquals(3600L,
tabletMetaInfo.getBinlogConfig().getTtlSeconds());
+ Assertions.assertEquals(4096L,
tabletMetaInfo.getBinlogConfig().getMaxBytes());
+ Assertions.assertEquals(7L,
tabletMetaInfo.getBinlogConfig().getMaxHistoryNums());
+
Assertions.assertEquals(org.apache.doris.proto.OlapFile.BinlogFormatPB.ROW,
+ tabletMetaInfo.getBinlogConfig().getBinlogFormat());
+
Assertions.assertTrue(tabletMetaInfo.getBinlogConfig().getNeedHistoricalValue());
+ }
+ }
+
+ @Test
+ public void testUpdateCompactionPolicyExcludesRowBinlogTablets() throws
Exception {
+ CloudSchemaChangeHandler handler = new CloudSchemaChangeHandler();
+ Database db = createMockDatabaseWithRowBinlogTablet();
+ Env env = Mockito.mock(Env.class);
+ MetaServiceProxy metaServiceProxy =
Mockito.mock(MetaServiceProxy.class);
+
Mockito.when(metaServiceProxy.updateTablet(Mockito.any())).thenReturn(okUpdateTabletResponse());
+
+ Config.enable_debug_points = true;
+
DebugPointUtil.addDebugPoint("CloudSchemaChangeHandler.notifyBackendsToSyncTabletMeta.skip");
+ try (MockedStatic<Env> envMock = Mockito.mockStatic(Env.class);
+ MockedStatic<MetaServiceProxy> metaProxyMock =
Mockito.mockStatic(MetaServiceProxy.class)) {
+ envMock.when(Env::getCurrentEnv).thenReturn(env);
+
metaProxyMock.when(MetaServiceProxy::getInstance).thenReturn(metaServiceProxy);
+
+ Map<String, String> properties = new HashMap<>();
+ properties.put(PropertyAnalyzer.PROPERTIES_COMPACTION_POLICY,
+ PropertyAnalyzer.TIME_SERIES_COMPACTION_POLICY);
+ handler.updateTableProperties(db, "tbl", properties);
+ }
+
+ ArgumentCaptor<Cloud.UpdateTabletRequest> updateCaptor =
+ ArgumentCaptor.forClass(Cloud.UpdateTabletRequest.class);
+ Mockito.verify(metaServiceProxy).updateTablet(updateCaptor.capture());
+ List<Cloud.TabletMetaInfoPB> tabletMetaInfos =
updateCaptor.getValue().getTabletMetaInfosList();
+ Assertions.assertEquals(Arrays.asList(101L), tabletMetaInfos.stream()
+
.map(Cloud.TabletMetaInfoPB::getTabletId).collect(Collectors.toList()));
+ Assertions.assertTrue(tabletMetaInfos.stream().allMatch(info ->
+
PropertyAnalyzer.TIME_SERIES_COMPACTION_POLICY.equals(info.getCompactionPolicy())));
+ }
+
+ @Test
+ public void testUpdateTtlIncludesRowBinlogTablets() throws Exception {
+ CloudSchemaChangeHandler handler = new CloudSchemaChangeHandler();
+ Database db = createMockDatabaseWithRowBinlogTablet();
+ Env env = Mockito.mock(Env.class);
+ MetaServiceProxy metaServiceProxy =
Mockito.mock(MetaServiceProxy.class);
+
Mockito.when(metaServiceProxy.updateTablet(Mockito.any())).thenReturn(okUpdateTabletResponse());
+
+ Config.enable_debug_points = true;
+
DebugPointUtil.addDebugPoint("CloudSchemaChangeHandler.notifyBackendsToSyncTabletMeta.skip");
+ try (MockedStatic<Env> envMock = Mockito.mockStatic(Env.class);
+ MockedStatic<MetaServiceProxy> metaProxyMock =
Mockito.mockStatic(MetaServiceProxy.class)) {
+ envMock.when(Env::getCurrentEnv).thenReturn(env);
+
metaProxyMock.when(MetaServiceProxy::getInstance).thenReturn(metaServiceProxy);
+
+ Map<String, String> properties = new HashMap<>();
+ properties.put(PropertyAnalyzer.PROPERTIES_FILE_CACHE_TTL_SECONDS,
"300");
+ handler.updateTableProperties(db, "tbl", properties);
+ }
+
+ ArgumentCaptor<Cloud.UpdateTabletRequest> updateCaptor =
+ ArgumentCaptor.forClass(Cloud.UpdateTabletRequest.class);
+ Mockito.verify(metaServiceProxy).updateTablet(updateCaptor.capture());
+ List<Cloud.TabletMetaInfoPB> tabletMetaInfos =
updateCaptor.getValue().getTabletMetaInfosList();
+ Assertions.assertEquals(Arrays.asList(101L, 201L),
tabletMetaInfos.stream()
+
.map(Cloud.TabletMetaInfoPB::getTabletId).collect(Collectors.toList()));
+ Assertions.assertTrue(tabletMetaInfos.stream().allMatch(info ->
info.getTtlSeconds() == 300));
+ }
+
@Test
public void testUpdateTablePropertiesThrowsWhenUpdateTabletFails() throws
Exception {
CloudSchemaChangeHandler handler = new CloudSchemaChangeHandler();
@@ -552,6 +651,8 @@ public class CloudSchemaChangeHandlerTest {
Mockito.when(partition.getName()).thenReturn("p1");
Mockito.when(partition.getMaterializedIndices(MaterializedIndex.IndexExtState.VISIBLE,
true))
.thenReturn(Arrays.asList(index));
+
Mockito.when(partition.getMaterializedIndices(MaterializedIndex.IndexExtState.VISIBLE,
false))
+ .thenReturn(Arrays.asList(index));
Mockito.when(index.getTablets()).thenReturn(Arrays.asList(tablet1,
tablet2, tablet3));
Mockito.when(tablet1.getId()).thenReturn(101L);
Mockito.when(tablet2.getId()).thenReturn(102L);
@@ -559,6 +660,33 @@ public class CloudSchemaChangeHandlerTest {
return db;
}
+ private Database createMockDatabaseWithRowBinlogTablet() throws Exception {
+ Database db = Mockito.mock(Database.class);
+ OlapTable table = Mockito.mock(OlapTable.class);
+ Partition partition = Mockito.mock(Partition.class);
+ MaterializedIndex baseIndex = Mockito.mock(MaterializedIndex.class);
+ MaterializedIndex rowBinlogIndex =
Mockito.mock(MaterializedIndex.class);
+ org.apache.doris.catalog.Tablet baseTablet =
Mockito.mock(org.apache.doris.catalog.Tablet.class);
+ org.apache.doris.catalog.Tablet rowBinlogTablet =
Mockito.mock(org.apache.doris.catalog.Tablet.class);
+
+ Mockito.when(db.getTableOrMetaException("tbl",
Table.TableType.OLAP)).thenReturn(table);
+ Mockito.when(table.getName()).thenReturn("tbl");
+
Mockito.when(table.getCompactionPolicy()).thenReturn(PropertyAnalyzer.SIZE_BASED_COMPACTION_POLICY);
+ Mockito.when(table.getKeysType()).thenReturn(KeysType.DUP_KEYS);
+
Mockito.when(table.getPartitions()).thenReturn(Arrays.asList(partition));
+ Mockito.when(table.getPartition("p1")).thenReturn(partition);
+ Mockito.when(partition.getName()).thenReturn("p1");
+
Mockito.when(partition.getMaterializedIndices(MaterializedIndex.IndexExtState.VISIBLE,
true))
+ .thenReturn(Arrays.asList(baseIndex, rowBinlogIndex));
+
Mockito.when(partition.getMaterializedIndices(MaterializedIndex.IndexExtState.VISIBLE,
false))
+ .thenReturn(Arrays.asList(baseIndex));
+
Mockito.when(baseIndex.getTablets()).thenReturn(Arrays.asList(baseTablet));
+
Mockito.when(rowBinlogIndex.getTablets()).thenReturn(Arrays.asList(rowBinlogTablet));
+ Mockito.when(baseTablet.getId()).thenReturn(101L);
+ Mockito.when(rowBinlogTablet.getId()).thenReturn(201L);
+ return db;
+ }
+
private InternalService.PSyncTabletMetaResponse okSyncTabletMetaResponse()
{
return InternalService.PSyncTabletMetaResponse.newBuilder()
.setStatus(org.apache.doris.proto.Types.PStatus.newBuilder()
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/CheckRowPolicyTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/CheckRowPolicyTest.java
index a67de805825..649bdd50f75 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/CheckRowPolicyTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/CheckRowPolicyTest.java
@@ -22,39 +22,58 @@ import org.apache.doris.analysis.UserDesc;
import org.apache.doris.analysis.UserIdentity;
import org.apache.doris.catalog.AccessPrivilege;
import org.apache.doris.catalog.AccessPrivilegeWithCols;
+import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.catalog.OlapTableWrapper;
+import org.apache.doris.common.Config;
import org.apache.doris.common.FeConstants;
import org.apache.doris.common.jmockit.Deencapsulation;
import org.apache.doris.mysql.privilege.AccessControllerManager;
import org.apache.doris.mysql.privilege.DataMaskPolicy;
+import org.apache.doris.nereids.CascadesContext;
import org.apache.doris.nereids.StatementContext;
import org.apache.doris.nereids.analyzer.UnboundRelation;
import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.properties.PhysicalProperties;
+import org.apache.doris.nereids.trees.expressions.Alias;
import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator;
+import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.commands.CreateUserCommand;
import
org.apache.doris.nereids.trees.plans.commands.GrantTablePrivilegeCommand;
import org.apache.doris.nereids.trees.plans.commands.info.CreateUserInfo;
import org.apache.doris.nereids.trees.plans.logical.LogicalCheckPolicy;
+import
org.apache.doris.nereids.trees.plans.logical.LogicalCheckPolicy.RelatedPolicy;
import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
+import org.apache.doris.nereids.util.PlanChecker;
import org.apache.doris.nereids.util.PlanRewriter;
import org.apache.doris.utframe.TestWithFeService;
import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
+import java.util.Locale;
import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
public class CheckRowPolicyTest extends TestWithFeService {
@@ -63,6 +82,7 @@ public class CheckRowPolicyTest extends TestWithFeService {
private static String tableName = "table1";
private static String tableNameRanddomDist = "tableRandomDist";
+ private static String tableNameMow = "tableMow";
private static String userName = "user1";
private static String policyName = "policy1";
@@ -71,6 +91,7 @@ public class CheckRowPolicyTest extends TestWithFeService {
@Override
protected void runBeforeAll() throws Exception {
FeConstants.runningUnitTest = true;
+ Config.enable_feature_binlog = true;
createDatabase(dbName);
useDatabase(dbName);
createTable("create table "
@@ -81,6 +102,13 @@ public class CheckRowPolicyTest extends TestWithFeService {
+ tableNameRanddomDist
+ " (k1 int, k2 int) AGGREGATE KEY(k1, k2) distributed by
random buckets 1"
+ " properties(\"replication_num\" = \"1\");");
+ createTable("create table "
+ + tableNameMow
+ + " (k1 int, k2 int) UNIQUE KEY(k1) distributed by hash(k1)
buckets 1"
+ + " properties(\"replication_num\" = \"1\","
+ + " \"enable_unique_key_merge_on_write\" = \"true\","
+ + " \"binlog.enable\" = \"true\", \"binlog.format\" = \"ROW\","
+ + " \"binlog.need_historical_value\" = \"true\");");
Database db =
Env.getCurrentInternalCatalog().getDbOrMetaException(fullDbName);
olapTable = (OlapTable) db.getTableOrAnalysisException(tableName);
@@ -104,19 +132,50 @@ public class CheckRowPolicyTest extends TestWithFeService
{
Mockito.doAnswer(invocation -> {
String tbl = invocation.getArgument(3);
String col = invocation.getArgument(4);
- return tbl.equalsIgnoreCase(tableNameRanddomDist)
- ? Optional.of(new DataMaskPolicy() {
- @Override
- public String getMaskTypeDef() {
- return String.format("concat(%s, '_****_', %s)",
col, col);
- }
-
- @Override
- public String getPolicyIdent() {
- return String.format("custom policy: concat(%s,
'_****_', %s)", col, col);
- }
- })
- : Optional.empty();
+ if (tbl.equalsIgnoreCase(tableNameRanddomDist)) {
+ return Optional.of(new DataMaskPolicy() {
+ @Override
+ public String getMaskTypeDef() {
+ return String.format("concat(%s, '_****_', %s)", col,
col);
+ }
+
+ @Override
+ public String getPolicyIdent() {
+ return String.format("custom policy: concat(%s,
'_****_', %s)", col, col);
+ }
+ });
+ }
+ if (!tbl.equalsIgnoreCase(tableNameMow)) {
+ return Optional.empty();
+ }
+ String column = col.toLowerCase(Locale.ROOT);
+ if (column.equalsIgnoreCase("k2")) {
+ String mask = "if(assert_true(k2 >= 0, 'post-snapshot row
reached mask'), k2, NULL)";
+ return Optional.of(new DataMaskPolicy() {
+ @Override
+ public String getMaskTypeDef() {
+ return mask;
+ }
+
+ @Override
+ public String getPolicyIdent() {
+ return "custom non-movable policy: " + mask;
+ }
+ });
+ }
+ // Mask hidden reconstruction columns too. Their aliases
deliberately get new ExprIds,
+ // so a reconstruction filter left above the mask would fail
CheckAfterRewrite.
+ return Optional.of(new DataMaskPolicy() {
+ @Override
+ public String getMaskTypeDef() {
+ return column;
+ }
+
+ @Override
+ public String getPolicyIdent() {
+ return "custom identity policy: " + column;
+ }
+ });
}).when(spyAcm).evalDataMaskPolicy(
Mockito.any(UserIdentity.class), Mockito.anyString(),
Mockito.anyString(), Mockito.anyString(), Mockito.anyString());
@@ -236,4 +295,102 @@ public class CheckRowPolicyTest extends TestWithFeService
{
+ " ON "
+ tableNameRanddomDist);
}
+
+ @Test
+ public void checkPolicyOnOlapTableWrapperUsesOriginTable() throws
Exception {
+ useUser(userName);
+ connectContext.setStatementContext(new StatementContext());
+ LogicalOlapScan relation = new
LogicalOlapScan(StatementScopeIdGenerator.newRelationId(),
+ new RenamedOlapTableWrapper(olapTable),
Arrays.asList(fullDbName));
+ LogicalCheckPolicy<LogicalOlapScan> checkPolicy = new
LogicalCheckPolicy<>(relation);
+ createPolicy("CREATE ROW POLICY " + policyName + " ON " + tableName
+ + " AS PERMISSIVE TO " + userName + " USING (k1 = 1)");
+ try {
+ Plan plan = PlanRewriter.bottomUpRewrite(checkPolicy,
connectContext,
+ new CheckPolicy(), new BindExpression());
+
+ Assertions.assertTrue(plan instanceof LogicalFilter);
+ LogicalFilter<?> filter = (LogicalFilter<?>) plan;
+ Assertions.assertEquals(relation, filter.child());
+
Assertions.assertTrue(filter.getConjuncts().toString().contains("k1"));
+ } finally {
+ dropPolicy("DROP ROW POLICY " + policyName + " ON " + tableName);
+ }
+ }
+
+ @Test
+ public void checkMvRefreshPolicyOnNestedOlapTableWrapperUsesOriginTable()
throws Exception {
+ useUser(userName);
+ StatementContext statementContext = new StatementContext();
+ connectContext.setStatementContext(statementContext);
+ OlapTableWrapper wrapper = new OlapTableWrapper(olapTable,
Collections.emptyMap());
+ OlapTableWrapper nestedWrapper = new OlapTableWrapper(wrapper,
Collections.emptyMap());
+ LogicalOlapScan relation = new
LogicalOlapScan(StatementScopeIdGenerator.newRelationId(),
+ nestedWrapper, Arrays.asList(fullDbName));
+ Expression predicate = new EqualTo(relation.getOutput().get(0), new
IntegerLiteral(1));
+ statementContext.setMvRefreshPredicates(ImmutableMap.of(olapTable,
ImmutableSet.of(predicate)));
+
+ LogicalCheckPolicy<LogicalOlapScan> checkPolicy = new
LogicalCheckPolicy<>(relation);
+ RelatedPolicy policy = checkPolicy.findPolicy(relation,
+ CascadesContext.initContext(statementContext, relation,
PhysicalProperties.GATHER));
+
+ Assertions.assertEquals(Optional.of(predicate),
policy.rowPolicyFilter);
+ }
+
+ @Test
+ public void mowTimeTravelReconstructionFiltersRunBeforeNonMovableMask()
throws Exception {
+ useUser(userName);
+ connectContext.getState().setIsQuery(true);
+
+ Plan rewrittenPlan = PlanChecker.from(connectContext)
+ .analyze("select k1, k2 from " + tableNameMow + " for version
as of 1001")
+ .rewrite()
+ .getPlan();
+ Set<LogicalUnion> unions = rewrittenPlan.collect(node -> node
instanceof LogicalUnion);
+ Assertions.assertEquals(1, unions.size());
+
+ LogicalUnion union = unions.iterator().next();
+ Assertions.assertEquals(2, union.children().size());
+ Set<String> reconstructionFilterSlots = new java.util.HashSet<>();
+ for (Plan branch : union.children()) {
+ Set<LogicalProject<?>> branchProjects = branch.collect(node ->
node instanceof LogicalProject);
+ int maskedK2ProjectCount = 0;
+ for (LogicalProject<?> project : branchProjects) {
+ for (NamedExpression namedExpression : project.getProjects()) {
+ if (namedExpression.getName().equalsIgnoreCase("k2")
+ && namedExpression instanceof Alias
+ && !(((Alias) namedExpression).child() instanceof
Slot)) {
+ maskedK2ProjectCount++;
+ }
+ }
+ }
+ Assertions.assertTrue(maskedK2ProjectCount >= 1,
+ "each MOW time-travel branch must retain its k2 data
mask");
+
+ Set<LogicalProject<?>> maskProjects = branch.collect(node -> node
instanceof LogicalProject
+ && ((LogicalProject<?>)
node).containsNoneMovableFunction());
+ Assertions.assertFalse(maskProjects.isEmpty(),
+ "each branch must contain the non-movable data mask");
+ for (LogicalProject<?> maskProject : maskProjects) {
+ Set<LogicalFilter<?>> filtersBelowMask =
maskProject.collect(node -> node instanceof LogicalFilter);
+ Assertions.assertFalse(filtersBelowMask.isEmpty(),
+ "reconstruction filters must run before every
non-movable data mask");
+ reconstructionFilterSlots.addAll(filtersBelowMask.stream()
+ .flatMap(filter -> filter.getConjuncts().stream())
+ .flatMap(conjunct -> conjunct.getInputSlots().stream())
+ .map(Slot::getName)
+ .collect(Collectors.toSet()));
+ }
+ }
+
Assertions.assertTrue(reconstructionFilterSlots.contains(Column.DELETE_SIGN));
+
Assertions.assertTrue(reconstructionFilterSlots.contains(Column.COMMIT_TSO_COL));
+
Assertions.assertTrue(reconstructionFilterSlots.contains(Column.BINLOG_OPERATION_COL));
+ }
+
+ private static class RenamedOlapTableWrapper extends OlapTableWrapper {
+ private RenamedOlapTableWrapper(OlapTable originTable) {
+ super(originTable, "renamed_policy_wrapper",
originTable.getBaseSchema(),
+ originTable.getKeysType(), Collections.emptyMap());
+ }
+ }
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/ExplainTableStreamPlanTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/ExplainTableStreamPlanTest.java
index e139ab4ab5e..06698b6b966 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/ExplainTableStreamPlanTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/ExplainTableStreamPlanTest.java
@@ -57,6 +57,7 @@ import
org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
import org.apache.doris.nereids.trees.plans.logical.LogicalOlapTableStreamScan;
import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
import org.apache.doris.nereids.trees.plans.physical.PhysicalPlan;
import org.apache.doris.nereids.util.MemoTestUtils;
import org.apache.doris.nereids.util.PlanChecker;
@@ -710,6 +711,29 @@ public class ExplainTableStreamPlanTest extends
TestWithFeService {
}
}
+ @Test
+ public void testMowTimeTravelBranchProjectionPreservesQualifier() {
+ Plan plan = PlanChecker.from(connectContext)
+ .analyze("select * from test_stream.tbl_stream_base for
version as of 1001")
+ .getCascadesContext().getRewritePlan();
+ Set<LogicalUnion> unions = plan.collect(node -> node instanceof
LogicalUnion);
+ Assertions.assertEquals(1, unions.size());
+
+ List<String> expectedQualifier = java.util.Arrays.asList(
+ "internal", "test_stream", "tbl_stream_base");
+ LogicalUnion union = unions.iterator().next();
+ Assertions.assertEquals(2, union.children().size());
+ for (Plan branch : union.children()) {
+ Assertions.assertInstanceOf(LogicalProject.class, branch);
+ LogicalProject<?> projection = (LogicalProject<?>) branch;
+ Assertions.assertFalse(projection.getProjects().isEmpty());
+ for (NamedExpression output : projection.getProjects()) {
+ Assertions.assertEquals(expectedQualifier,
output.getQualifier(),
+ "MOW time-travel branch aliases must keep the scan
qualifier");
+ }
+ }
+ }
+
@Test
public void testMowTimeTravelQualifiedColumnCanBind() {
// MOW time-travel goes through a union whose outputs are rebuilt with
empty qualifiers.
diff --git a/gensrc/proto/cloud.proto b/gensrc/proto/cloud.proto
index 62e94a9b75a..a6cfecdcb85 100644
--- a/gensrc/proto/cloud.proto
+++ b/gensrc/proto/cloud.proto
@@ -736,6 +736,7 @@ message TabletMetaInfoPB { // For update tablet meta
optional bool enable_mow_light_delete = 14;
optional int32 vertical_compaction_num_columns_per_group = 15;
optional string group_commit_mode = 16;
+ optional doris.BinlogConfigPB binlog_config = 17;
}
message TabletCompactionJobPB {
diff --git
a/regression-test/data/row_binlog_p0/test_row_binlog_mow_light_delete.out
b/regression-test/data/row_binlog_p0/test_row_binlog_mow_light_delete.out
new file mode 100644
index 00000000000..5b12cd39019
--- /dev/null
+++ b/regression-test/data/row_binlog_p0/test_row_binlog_mow_light_delete.out
@@ -0,0 +1,16 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !without_history_base --
+2 two
+
+-- !without_history_binlog --
+0 1 one
+0 2 two
+2 1 \N
+
+-- !with_history_base --
+2 two
+
+-- !with_history_binlog --
+0 1 one \N
+0 2 two \N
+2 1 one one
diff --git
a/regression-test/data/time_travel_p0/test_mow_time_travel_row_policy.out
b/regression-test/data/time_travel_p0/test_mow_time_travel_row_policy.out
new file mode 100644
index 00000000000..fab45bf8500
--- /dev/null
+++ b/regression-test/data/time_travel_p0/test_mow_time_travel_row_policy.out
@@ -0,0 +1,15 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !latest_image --
+1 10
+
+-- !historical_image --
+1 10
+3 30
+
+-- !historical_dotted_column_unqualified --
+100
+300
+
+-- !historical_dotted_column_qualified --
+100
+300
diff --git
a/regression-test/suites/row_binlog_p0/test_row_binlog_mow_light_delete.groovy
b/regression-test/suites/row_binlog_p0/test_row_binlog_mow_light_delete.groovy
new file mode 100644
index 00000000000..096e0fdee65
--- /dev/null
+++
b/regression-test/suites/row_binlog_p0/test_row_binlog_mow_light_delete.groovy
@@ -0,0 +1,104 @@
+// 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.
+
+suite("test_row_binlog_mow_light_delete", "nonConcurrent") {
+ sql "DROP TABLE IF EXISTS test_row_binlog_mow_light_delete_without_history
FORCE"
+ sql "DROP TABLE IF EXISTS test_row_binlog_mow_light_delete_with_history
FORCE"
+ sql "DROP TABLE IF EXISTS test_row_binlog_dup_delete FORCE"
+
+ sql """
+ CREATE TABLE test_row_binlog_mow_light_delete_without_history (
+ k INT,
+ v STRING
+ )
+ UNIQUE KEY(k)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "enable_mow_light_delete" = "true",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW",
+ "binlog.need_historical_value" = "false"
+ )
+ """
+
+ sql "INSERT INTO test_row_binlog_mow_light_delete_without_history VALUES
(1, 'one'), (2, 'two')"
+ sql "DELETE FROM test_row_binlog_mow_light_delete_without_history WHERE k
= 1"
+
+ order_qt_without_history_base """
+ SELECT k, v
+ FROM test_row_binlog_mow_light_delete_without_history
+ """
+
+ qt_without_history_binlog """
+ SELECT __DORIS_BINLOG_OP__ AS op, k, v
+ FROM binlog("table" =
"test_row_binlog_mow_light_delete_without_history")
+ ORDER BY __DORIS_BINLOG_TSO__, __DORIS_BINLOG_LSN__
+ """
+
+ sql """
+ CREATE TABLE test_row_binlog_mow_light_delete_with_history (
+ k INT,
+ v STRING
+ )
+ UNIQUE KEY(k)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "enable_mow_light_delete" = "true",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW",
+ "binlog.need_historical_value" = "true"
+ )
+ """
+
+ sql "INSERT INTO test_row_binlog_mow_light_delete_with_history VALUES (1,
'one'), (2, 'two')"
+ sql "DELETE FROM test_row_binlog_mow_light_delete_with_history WHERE k = 1"
+
+ order_qt_with_history_base """
+ SELECT k, v
+ FROM test_row_binlog_mow_light_delete_with_history
+ """
+
+ qt_with_history_binlog """
+ SELECT __DORIS_BINLOG_OP__ AS op, k, v, __BEFORE__v__
+ FROM binlog("table" = "test_row_binlog_mow_light_delete_with_history")
+ ORDER BY __DORIS_BINLOG_TSO__, __DORIS_BINLOG_LSN__
+ """
+
+ sql """
+ CREATE TABLE test_row_binlog_dup_delete (
+ k INT,
+ v STRING
+ )
+ DUPLICATE KEY(k)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW"
+ )
+ """
+
+ sql "INSERT INTO test_row_binlog_dup_delete VALUES (1, 'one')"
+ test {
+ sql "DELETE FROM test_row_binlog_dup_delete WHERE k = 1"
+ exception "DELETE with predicates is not supported when binlog<row> is
enabled"
+ }
+}
diff --git
a/regression-test/suites/time_travel_p0/test_mow_time_travel_row_policy.groovy
b/regression-test/suites/time_travel_p0/test_mow_time_travel_row_policy.groovy
new file mode 100644
index 00000000000..275b20f04c9
--- /dev/null
+++
b/regression-test/suites/time_travel_p0/test_mow_time_travel_row_policy.groovy
@@ -0,0 +1,107 @@
+// 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.
+
+suite("test_mow_time_travel_row_policy", "nonConcurrent,p0,auth") {
+ if (isCloudMode()) {
+ return
+ }
+
+ String dbName = "test_mow_time_travel_row_policy_db"
+ String tableName = "test_mow_time_travel_row_policy_table"
+ String user = "test_mow_time_travel_row_policy_user"
+ String password = "C123_567p"
+ String restrictivePolicy = "test_mow_time_travel_value_policy"
+ String firstKeyPolicy = "test_mow_time_travel_first_key_policy"
+ String thirdKeyPolicy = "test_mow_time_travel_third_key_policy"
+
+ try_sql "DROP ROW POLICY IF EXISTS ${restrictivePolicy} ON
${dbName}.${tableName} FOR ${user}"
+ try_sql "DROP ROW POLICY IF EXISTS ${firstKeyPolicy} ON
${dbName}.${tableName} FOR ${user}"
+ try_sql "DROP ROW POLICY IF EXISTS ${thirdKeyPolicy} ON
${dbName}.${tableName} FOR ${user}"
+ try_sql "DROP USER IF EXISTS ${user}"
+ sql "DROP DATABASE IF EXISTS ${dbName}"
+
+ sql "CREATE DATABASE ${dbName}"
+ sql """
+ CREATE TABLE ${dbName}.${tableName} (
+ k INT,
+ v INT,
+ `v1.v2` INT
+ )
+ UNIQUE KEY(k)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES (
+ 'replication_num' = '1',
+ 'enable_unique_key_merge_on_write' = 'true',
+ 'binlog.enable' = 'true',
+ 'binlog.format' = 'ROW',
+ 'binlog.need_historical_value' = 'true'
+ )
+ """
+
+ sql "INSERT INTO ${dbName}.${tableName} VALUES (1, 10, 100), (2, 20, 200),
(3, 30, 300), (5, 50, 500)"
+ sql "SET show_hidden_columns = true"
+ long snapshotTso = sql("SELECT MAX(__DORIS_COMMIT_TSO_COL__) FROM
${dbName}.${tableName}")[0][0] as Long
+ sql "SET show_hidden_columns = false"
+
+ // The historical left branch contains unchanged keys 1 and 5. The
row-binlog right branch
+ // restores updated key 2 and deleted key 3. A new key 4 must not appear
in the snapshot.
+ sql "INSERT INTO ${dbName}.${tableName} VALUES (2, 200, 2000)"
+ sql "DELETE FROM ${dbName}.${tableName} WHERE k = 3"
+ sql "INSERT INTO ${dbName}.${tableName} VALUES (4, 40, 400)"
+
+ sql "CREATE USER '${user}' IDENTIFIED BY '${password}'"
+ sql "GRANT SELECT_PRIV ON internal.${dbName}.${tableName} TO ${user}"
+ sql """
+ CREATE ROW POLICY ${restrictivePolicy} ON ${dbName}.${tableName}
+ AS RESTRICTIVE TO ${user} USING (v <= 30)
+ """
+ sql """
+ CREATE ROW POLICY ${firstKeyPolicy} ON ${dbName}.${tableName}
+ AS PERMISSIVE TO ${user} USING (k = 1)
+ """
+ sql """
+ CREATE ROW POLICY ${thirdKeyPolicy} ON ${dbName}.${tableName}
+ AS PERMISSIVE TO ${user} USING (k = 3)
+ """
+
+ String userJdbcUrl = org.apache.doris.regression.Config.buildUrlWithDb(
+ context.config.jdbcUrl, dbName)
+ connect(user, password, userJdbcUrl) {
+ sql "SET enable_nereids_planner = true"
+ sql "SET enable_fallback_to_original_planner = false"
+ sql "SET enable_sql_cache = false"
+
+ // Restrictive AND (permissive OR permissive) leaves only key 1 in the
latest image.
+ order_qt_latest_image "SELECT k, v FROM ${tableName} ORDER BY k"
+
+ // key 1 is supplied by the base-scan branch and key 3 by the
row-binlog branch. A
+ // missing policy marker on either scan leaks key 5 or key 2
respectively.
+ order_qt_historical_image """
+ SELECT k, v FROM ${tableName} FOR VERSION AS OF ${snapshotTso}
ORDER BY k
+ """
+
+ // Keep the raw name of a dotted primitive column through both union
branches. Exercise
+ // unqualified and table-qualified binding because
UnboundSlot.getName() renders backticks.
+ order_qt_historical_dotted_column_unqualified """
+ SELECT `v1.v2` FROM ${tableName} FOR VERSION AS OF ${snapshotTso}
ORDER BY k
+ """
+ order_qt_historical_dotted_column_qualified """
+ SELECT ${tableName}.`v1.v2` FROM ${tableName} FOR VERSION AS OF
${snapshotTso}
+ ORDER BY ${tableName}.k
+ """
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
