morrySnow commented on code in PR #64776:
URL: https://github.com/apache/doris/pull/64776#discussion_r3489073665
##########
regression-test/suites/table_stream_p0/test_min_delta_stream.groovy:
##########
@@ -837,6 +830,6 @@ suite("test_min_delta_stream", "nonConcurrent") {
exception "MIN_DELTA INCR query requires base table to enable
binlog.need_historical_value=true"
}
} finally {
- sql "DROP DATABASE IF EXISTS test_min_delta_stream_db"
+ // sql "DROP DATABASE IF EXISTS test_min_delta_stream_db"
}
Review Comment:
⚠️ **Regression test cleanup is commented out.**
This will leave the test database `test_min_delta_stream_db` on disk after
the test runs, which can pollute subsequent test executions and waste storage.
Please uncomment the cleanup or explain why the database should not be
dropped.
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTableWrapper.java:
##########
@@ -154,4 +168,15 @@ public MaterializedIndexMeta getBaseIndexMeta() {
public MaterializedIndex getBaseIndex() {
return originTable.getBaseIndex();
}
+
+ public Pair<Long, Long> getPartitionOffset(long partitionId) {
+ return partitionOffsetMap.get(partitionId);
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ return super.equals(obj)
+ && originTable.equals(((OlapTableWrapper) obj).originTable)
+ && partitionOffsetMap.equals(((OlapTableWrapper)
obj).partitionOffsetMap);
Review Comment:
⚠️ **`equals()` is missing an `instanceof` guard.**
If `obj` is a non-wrapper `OlapTable` (or any other type) and
`super.equals(obj)` returns `true` (possible since `OlapTableWrapper` extends
`OlapTable`), the cast to `OlapTableWrapper` will throw a `ClassCastException`.
Add an `instanceof` check before the cast, e.g.:
```java
if (!(obj instanceof OlapTableWrapper)) return false;
OlapTableWrapper that = (OlapTableWrapper) obj;
return super.equals(obj)
&& originTable.equals(that.originTable)
&& partitionOffsetMap.equals(that.partitionOffsetMap);
```
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/RowBinlogTableWrapper.java:
##########
@@ -71,12 +67,14 @@ public MaterializedIndex getPartitionIndex(Partition
partition, long indexId) {
return null;
}
- public Optional<OlapTableStreamWrapper> getParent() {
- return parent;
- }
-
@Override
public KeysType getKeysType() {
return KeysType.DUP_KEYS;
}
+
+ @Override
+ public boolean equals(Object obj) {
+ return super.equals(obj)
+ && rowBinlogMeta.equals(((RowBinlogTableWrapper)
obj).rowBinlogMeta);
+ }
Review Comment:
⚠️ **`equals()` is missing an `instanceof` guard.**
Same issue as `OlapTableWrapper.equals()` — if `obj` is not a
`RowBinlogTableWrapper` but `super.equals()` returns `true`, the cast throws
`ClassCastException`.
Add the same `instanceof` guard pattern here.
##########
fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java:
##########
@@ -546,11 +540,12 @@ private void addScanRangeLocations(Partition partition,
);
paloRange.setVersionHash("");
paloRange.setTabletId(tabletId);
- if (olapTable instanceof RowBinlogTableWrapper) {
+ if (olapTable instanceof OlapTableWrapper) {
TBinlogScanType binlogScanType =
- parseBinlogScanType(scanParams,
((RowBinlogTableWrapper) olapTable).getOriginTable());
- if (((RowBinlogTableWrapper)
olapTable).getParent().isPresent()) {
- Pair<Long, Long> update =
getStreamUpdate(partition.getId());
+ parseBinlogScanType(scanParams, ((OlapTableWrapper)
olapTable).getOriginTable());
+ Pair<Long, Long> update =
getPartitionOffset(partition.getId());
+ if (update != null) {
+ context.getSessionVariable().setEnableParallelScan(false);
if (update.first != null) {
Review Comment:
⚠️ **Mutating a session variable inside `addScanRangeLocations()` is a
hidden side effect.**
`setEnableParallelScan(false)` modifies the session variable from within a
scan node's initialization path. This is surprising because:
1. It couples scan range logic to session state management
2. It could affect other parts of the query that check `enableParallelScan`
after this point
3. If `context.getSessionVariable()` is ever null from a non-standard code
path, this will NPE
Consider passing this as a flag on the scan node or scan ranges instead of
mutating the session.
##########
be/src/storage/segment/column_reader_cache.cpp:
##########
@@ -119,6 +120,7 @@ Status ColumnReaderCache::get_column_reader(int32_t col_uid,
ColumnReaderOptions opts {.kept_in_memory = _tablet_schema->is_in_memory(),
.be_exec_version = _be_exec_version,
.tablet_schema = _tablet_schema};
+ opts.const_value = std::move(const_value);
Review Comment:
💡 **Consider adding a comment about cache key safety.**
`_lookup({col_uid, {}})` uses only `col_uid` as the cache key, while
`const_value` is passed separately and only used when creating a new reader.
This is safe today because `const_value` is invariant per segment (derived from
version + commit_tso, both segment-level properties).
However, this invariant is non-obvious — a future change that varies
`const_value` per-query (e.g., time-travel at different snapshots) would
produce stale cached values. Add a comment here explaining why the cache key
doesn't need to include `const_value`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
