github-actions[bot] commented on code in PR #69850: URL: https://github.com/apache/doris/pull/69850#discussion_r4424439892
########## fe/fe-core/src/main/java/org/apache/doris/catalog/stream/PartitionStreamState.java: ########## @@ -0,0 +1,131 @@ +// 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; + + @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; + } + } + return Math.max(1, delta); Review Comment: [P1] Keep a known nonempty changelog range from collapsing to one row when counters are unavailable. `estimatePartitionChangelogRead()` can pass a positive `(startTso,endTso]` while both tablet counters and `getBaseTableRowCount()` are still zero (for example, a large load immediately after the stream offset, before backend reports or analysis stats arrive). `changelogDelta()` then returns UNKNOWN, the fallback leaves `delta` at zero, and `Math.max(1, delta)` reports one row even though the changelog scan reads the entire interval; the optimizer can choose that large stream as a broadcast build and exhaust memory. Preserve an unknown/conservative estimate or defer the decision when the interval is known but every counter is unavailable, and add an integration case for planning before tablet statistics are published. -- 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]
