github-actions[bot] commented on code in PR #69850: URL: https://github.com/apache/doris/pull/69850#discussion_r4424820835
########## fe/fe-core/src/main/java/org/apache/doris/catalog/stream/PartitionStreamState.java: ########## @@ -0,0 +1,146 @@ +// 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.catalog.stream; + +import com.google.gson.annotations.SerializedName; + +/** + * What a table stream recorded for one partition when it last advanced that partition's offset. + * + * <p>A changelog read returns the change rows between the stream offset and the end TSO of the + * read, a number the FE cannot derive from table metadata: the changelog index of a partition + * holds the change rows of every offset range, so its row count is the size of the whole + * changelog, not of the range being read. Two counters recorded next to the offset give the + * estimate a baseline to subtract: + * + * <ul> + * <li>{@code changelogRowCount} (P): the row count of the partition's changelog index at the + * moment the offset advanced; the read estimate is the current row count minus it;</li> + * <li>{@code baseRowCount} (Q): the row count of the partition's base index at the same moment; + * a snapshot read answers with the image at the offset, and Q is its size.</li> + * </ul> + * + * <p>The rule behind the estimates is to err on the large side: over-estimating a relation only + * makes the optimizer plan it differently, while under-estimating lets a large relation become + * the build side of a broadcast join and exhaust memory. + */ +public class PartitionStreamState { + public static final long UNKNOWN_ROW_COUNT = -1L; + + /** + * The estimate of a read of a range which is known to hold changes when no counter can size + * it: the backends have not reported the partition, a load just wrote it, and neither its + * changelog index nor its base index has a count the frontend can use. Reporting one row there + * would let the optimizer take a range of unknown size for a small relation, which is the + * failure this estimate exists to avoid, so the read is reported as a relation too large to be + * broadcast instead. + */ + public static final long UNKNOWN_NONEMPTY_ROW_COUNT = 1L << 30; + + @SerializedName("crc") + private final long changelogRowCount; + + @SerializedName("brc") + private final long baseRowCount; + + public PartitionStreamState(long changelogRowCount, long baseRowCount) { + this.changelogRowCount = changelogRowCount; + this.baseRowCount = baseRowCount; + } + + public long getChangelogRowCount() { + return changelogRowCount; + } + + public long getBaseRowCount() { + return baseRowCount; + } + + /** + * Change rows the read of {@code (startTso, endTso]} may return, or {@link #UNKNOWN_ROW_COUNT} + * when no counter can tell. + * + * <p>A read of an empty range returns nothing no matter what the row counts report: the two + * TSOs are commit metadata that do not lag behind, unlike the row counts, so they decide the + * empty case. Otherwise the rows still in the changelog bound the read: a valid read only + * returns rows which were not reclaimed yet. + */ + public static long changelogDelta(long startTso, long endTso, long storedChangelogRowCount, + long currentChangelogRowCount) { + if (startTso < 0 || endTso < 0) { + // The range of the read is not known, so the difference of the counters cannot be + // attributed to it: the read may start before the counters do. + return UNKNOWN_ROW_COUNT; + } + if (endTso <= startTso) { + return 0; + } + if (storedChangelogRowCount >= 0 && currentChangelogRowCount > storedChangelogRowCount) { Review Comment: [P1] Keep the delta conservative across partial TTL reclamation. The baseline is recorded before old changelog rowsets are removed, while `applyChangelogTtl()` clips the scan start to the first retained TSO. If the baseline is 100, TTL purges 90 old rows, and 110 new rows are appended, the current count is 120 and this branch returns 20 even though the clipped range can contain 110 rows. That under-estimate can make a large range a broadcast build. Detect a reclaimed/TTL-clipped baseline or use a retained-count bound, and add a purge-then-growth test. ########## fe/fe-core/src/main/java/org/apache/doris/catalog/stream/PartitionStreamState.java: ########## @@ -0,0 +1,146 @@ +// 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.catalog.stream; + +import com.google.gson.annotations.SerializedName; + +/** + * What a table stream recorded for one partition when it last advanced that partition's offset. + * + * <p>A changelog read returns the change rows between the stream offset and the end TSO of the + * read, a number the FE cannot derive from table metadata: the changelog index of a partition + * holds the change rows of every offset range, so its row count is the size of the whole + * changelog, not of the range being read. Two counters recorded next to the offset give the + * estimate a baseline to subtract: + * + * <ul> + * <li>{@code changelogRowCount} (P): the row count of the partition's changelog index at the + * moment the offset advanced; the read estimate is the current row count minus it;</li> + * <li>{@code baseRowCount} (Q): the row count of the partition's base index at the same moment; + * a snapshot read answers with the image at the offset, and Q is its size.</li> + * </ul> + * + * <p>The rule behind the estimates is to err on the large side: over-estimating a relation only + * makes the optimizer plan it differently, while under-estimating lets a large relation become + * the build side of a broadcast join and exhaust memory. + */ +public class PartitionStreamState { + public static final long UNKNOWN_ROW_COUNT = -1L; + + /** + * The estimate of a read of a range which is known to hold changes when no counter can size + * it: the backends have not reported the partition, a load just wrote it, and neither its + * changelog index nor its base index has a count the frontend can use. Reporting one row there + * would let the optimizer take a range of unknown size for a small relation, which is the + * failure this estimate exists to avoid, so the read is reported as a relation too large to be + * broadcast instead. + */ + public static final long UNKNOWN_NONEMPTY_ROW_COUNT = 1L << 30; + + @SerializedName("crc") + private final long changelogRowCount; + + @SerializedName("brc") + private final long baseRowCount; + + public PartitionStreamState(long changelogRowCount, long baseRowCount) { + this.changelogRowCount = changelogRowCount; + this.baseRowCount = baseRowCount; + } + + public long getChangelogRowCount() { + return changelogRowCount; + } + + public long getBaseRowCount() { + return baseRowCount; + } + + /** + * Change rows the read of {@code (startTso, endTso]} may return, or {@link #UNKNOWN_ROW_COUNT} + * when no counter can tell. + * + * <p>A read of an empty range returns nothing no matter what the row counts report: the two + * TSOs are commit metadata that do not lag behind, unlike the row counts, so they decide the + * empty case. Otherwise the rows still in the changelog bound the read: a valid read only + * returns rows which were not reclaimed yet. + */ + public static long changelogDelta(long startTso, long endTso, long storedChangelogRowCount, + long currentChangelogRowCount) { + if (startTso < 0 || endTso < 0) { + // The range of the read is not known, so the difference of the counters cannot be + // attributed to it: the read may start before the counters do. + return UNKNOWN_ROW_COUNT; + } + if (endTso <= startTso) { + return 0; + } + if (storedChangelogRowCount >= 0 && currentChangelogRowCount > storedChangelogRowCount) { + return currentChangelogRowCount - storedChangelogRowCount; + } + if (currentChangelogRowCount > 0) { + // No baseline, or the counters moved backwards (a reclamation, or a partition which + // was recreated): every row the read may return is in the changelog right now. + return currentChangelogRowCount; + } + return UNKNOWN_ROW_COUNT; + } + + /** + * Estimate of a changelog read. See {@link #changelogDelta}. When the row counts cannot bound + * the read (not reported yet, or the partition was just written and the backend has not + * reported it), fall back to values which keep the estimate in the right order of magnitude + * instead of returning one row: a base row can be changed many times, so the base row count + * is not a strict bound, it only keeps the relation from looking tiny. + */ + public static long estimateChangelogRead(long startTso, long endTso, long storedChangelogRowCount, + long currentChangelogRowCount, long currentBaseRowCount, long tableRowCount) { + long delta = changelogDelta(startTso, endTso, storedChangelogRowCount, currentChangelogRowCount); + if (delta < 0) { + delta = Math.max(currentChangelogRowCount, currentBaseRowCount); + if (delta <= 0) { + delta = tableRowCount; + } + if (delta <= 0 && startTso >= 0 && endTso > startTso) { Review Comment: [P1] Do not treat an empty-at-create partition's first load as a one-row range. Its stored offset is -1, which `toExclusiveBound` converts to an unset start; once the partition receives a large first load, `filterIncrementalPartitionIds()` still plans the changelog scan from the beginning. With counters and table stats still 0, this `startTso >= 0` guard is false and `Math.max(1, delta)` reports one row, so the full first-load range can become a broadcast build. Treat an unknown start with a known nonempty/end range conservatively and add this case to the regression coverage. ########## fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java: ########## @@ -835,6 +837,7 @@ private void executeCommitTxnRequest(CommitTxnRequest.Builder builder, List<Tabl try { txnState = commitTxn(builder, tableList, transactionId, is2PC, tabletCommitInfos, tabletIds); txnOperated = true; + recordStreamPartitionRowCounts(builder.getTableStreamUpdatesList()); Review Comment: [P1] Skip baseline recording when a cloud commit is only an idempotent retry. After a persisted transaction is already visible, MetaService returns before applying the table-stream update, but `commitTxn()` accepts that response and this line records the builder's partitions anyway. Rows/reporting that arrive between the original commit and the retry are then stored as the baseline for the unchanged offset, so the next `current - stored` estimate omits them and can choose a broadcast build. Record only on the transition or make the update keyed/idempotent, and add a duplicate-commit test. ########## fe/fe-core/src/main/java/org/apache/doris/catalog/stream/OlapTableStreamWrapper.java: ########## @@ -103,15 +130,21 @@ public void installCloudReadStates(Map<Long, Cloud.TableStreamPartitionReadState "Cloud Table Stream offset exceeds the statement end TSO"); } } - cloudReadStates = immutableReadStates; - cloudReadStatesInstalled = true; - outputUpdateMap.clear(); - cloudReadStates.forEach((partitionId, state) -> { - if (state.getVisibleVersion() > Partition.PARTITION_INIT_VERSION) { - Long startTso = state.hasOffsetTso() ? state.getOffsetTso() : null; - outputUpdateMap.put(partitionId, Pair.of(startTso, state.getEndTso())); - } - }); + stream.readLock(); + try { + cloudReadStates = immutableReadStates; + cloudReadStatesInstalled = true; + outputUpdateMap.clear(); + cloudReadStates.forEach((partitionId, state) -> { + if (state.getVisibleVersion() > Partition.PARTITION_INIT_VERSION) { + Long startTso = state.hasOffsetTso() ? state.getOffsetTso() : null; + outputUpdateMap.put(partitionId, Pair.of(startTso, state.getEndTso())); + } + }); + partitionStreamStates = snapshotPartitionStreamStates(outputUpdateMap.keySet()); Review Comment: [P1] Fence the cloud read-state snapshot with the local baseline. `CloudTableStreamReadStateHook` fetches the authoritative range from MetaService before calling this method; a concurrent consumer can advance the stream and replace `partitionStates` in that gap. This then snapshots the newer baseline under the lock while installing the older `(offsetTso,endTso)` range, so `current - stored` can collapse a large cloud read and choose a broadcast build (the IVM rewrite path calls the hook outside a statement-wide stream lock). Carry a generation/offset token through installation or retry when local state advanced, and add a concurrent cloud plan/consume test. ########## fe/fe-core/src/main/java/org/apache/doris/catalog/stream/OlapTableStream.java: ########## @@ -318,6 +385,7 @@ public void unprotectedUpdateStreamUpdate(AbstractTableStreamUpdate update, Long partitionOffset.put(entry.getKey(), entry.getValue()); partitionConsumptionTime.put(entry.getKey(), ts); } + unprotectedRecordPartitionStreamStates(next.keySet()); Review Comment: [P1] Do not sample counters after advancing the stream offset without a source-TSO fence. The consumer commit locks the sink and stream, but not the stream's base table; a producer can commit/report rows with TSO greater than the captured `next` between the offset write and this call. The stored baseline then already includes rows from the next `(startTso,endTso]` range, so `current - stored` can be zero/one and select a large changelog as a broadcast build. Capture counters at the offset boundary or version/filter them by source TSO, and add a producer/consumer race test. ########## fe/fe-core/src/main/java/org/apache/doris/catalog/stream/PartitionStreamState.java: ########## @@ -0,0 +1,146 @@ +// 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.catalog.stream; + +import com.google.gson.annotations.SerializedName; + +/** + * What a table stream recorded for one partition when it last advanced that partition's offset. + * + * <p>A changelog read returns the change rows between the stream offset and the end TSO of the + * read, a number the FE cannot derive from table metadata: the changelog index of a partition + * holds the change rows of every offset range, so its row count is the size of the whole + * changelog, not of the range being read. Two counters recorded next to the offset give the + * estimate a baseline to subtract: + * + * <ul> + * <li>{@code changelogRowCount} (P): the row count of the partition's changelog index at the + * moment the offset advanced; the read estimate is the current row count minus it;</li> + * <li>{@code baseRowCount} (Q): the row count of the partition's base index at the same moment; + * a snapshot read answers with the image at the offset, and Q is its size.</li> + * </ul> + * + * <p>The rule behind the estimates is to err on the large side: over-estimating a relation only + * makes the optimizer plan it differently, while under-estimating lets a large relation become + * the build side of a broadcast join and exhaust memory. + */ +public class PartitionStreamState { + public static final long UNKNOWN_ROW_COUNT = -1L; + + /** + * The estimate of a read of a range which is known to hold changes when no counter can size + * it: the backends have not reported the partition, a load just wrote it, and neither its + * changelog index nor its base index has a count the frontend can use. Reporting one row there + * would let the optimizer take a range of unknown size for a small relation, which is the + * failure this estimate exists to avoid, so the read is reported as a relation too large to be + * broadcast instead. + */ + public static final long UNKNOWN_NONEMPTY_ROW_COUNT = 1L << 30; + + @SerializedName("crc") + private final long changelogRowCount; + + @SerializedName("brc") + private final long baseRowCount; + + public PartitionStreamState(long changelogRowCount, long baseRowCount) { + this.changelogRowCount = changelogRowCount; + this.baseRowCount = baseRowCount; + } + + public long getChangelogRowCount() { + return changelogRowCount; + } + + public long getBaseRowCount() { + return baseRowCount; + } + + /** + * Change rows the read of {@code (startTso, endTso]} may return, or {@link #UNKNOWN_ROW_COUNT} + * when no counter can tell. + * + * <p>A read of an empty range returns nothing no matter what the row counts report: the two + * TSOs are commit metadata that do not lag behind, unlike the row counts, so they decide the + * empty case. Otherwise the rows still in the changelog bound the read: a valid read only + * returns rows which were not reclaimed yet. + */ + public static long changelogDelta(long startTso, long endTso, long storedChangelogRowCount, + long currentChangelogRowCount) { + if (startTso < 0 || endTso < 0) { + // The range of the read is not known, so the difference of the counters cannot be + // attributed to it: the read may start before the counters do. + return UNKNOWN_ROW_COUNT; + } + if (endTso <= startTso) { + return 0; + } + if (storedChangelogRowCount >= 0 && currentChangelogRowCount > storedChangelogRowCount) { + return currentChangelogRowCount - storedChangelogRowCount; + } + if (currentChangelogRowCount > 0) { + // No baseline, or the counters moved backwards (a reclamation, or a partition which + // was recreated): every row the read may return is in the changelog right now. + return currentChangelogRowCount; + } + return UNKNOWN_ROW_COUNT; + } + + /** + * Estimate of a changelog read. See {@link #changelogDelta}. When the row counts cannot bound + * the read (not reported yet, or the partition was just written and the backend has not + * reported it), fall back to values which keep the estimate in the right order of magnitude + * instead of returning one row: a base row can be changed many times, so the base row count + * is not a strict bound, it only keeps the relation from looking tiny. + */ + public static long estimateChangelogRead(long startTso, long endTso, long storedChangelogRowCount, + long currentChangelogRowCount, long currentBaseRowCount, long tableRowCount) { + long delta = changelogDelta(startTso, endTso, storedChangelogRowCount, currentChangelogRowCount); + if (delta < 0) { + delta = Math.max(currentChangelogRowCount, currentBaseRowCount); + if (delta <= 0) { + delta = tableRowCount; + } + if (delta <= 0 && startTso >= 0 && endTso > startTso) { + // The TSOs say the range holds changes, but nothing sizes them: a load which just + // wrote the partition, before any counter exists. One row would say the opposite. + delta = UNKNOWN_NONEMPTY_ROW_COUNT; + } + } + return Math.max(1, delta); + } + + /** + * Estimate of a snapshot read answering with the image at the stream offset. + * + * <p>For a duplicate key table the image only grows by appends, so the current base rows are an + * upper bound and adding the change rows would only make the estimate worse. For a unique or + * aggregating table a delete after the offset makes the image larger than the current rows, + * and the rows added back never exceed the change rows of the range. + */ + public static long estimateSnapshotRead(boolean appendOnly, long storedBaseRowCount, + long currentBaseRowCount, long changelogDelta, long tableRowCount) { + long currentRows = Math.max(0, currentBaseRowCount); + long addedBack = appendOnly || changelogDelta < 0 ? 0 : changelogDelta; + long estimate = Math.max(storedBaseRowCount, currentRows + addedBack); + if (estimate <= 0) { + estimate = tableRowCount; + } + return Math.max(1, estimate); Review Comment: [P1] Keep an unknown snapshot from becoming a one-row build side. For a partition with a positive stream offset and real rows, `initializeLocalOffsets()` and `getBaseTableRowCount()` can both see unreported counters as 0, so `estimateSnapshotRead(true, 0, 0, UNKNOWN, 0)` reaches this `Math.max(1, estimate)` and returns 1. The snapshot wrapper then overrides the delegated table count with that one-row estimate, allowing a large image to be chosen as a broadcast build. Preserve a conservative unknown/nonempty estimate when all counters are unavailable and add a pre-report snapshot-planning case. -- 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]
