github-actions[bot] commented on code in PR #68713: URL: https://github.com/apache/doris/pull/68713#discussion_r4390052446
########## fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonJniHeapEstimate.java: ########## @@ -0,0 +1,234 @@ +// 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.connector.paimon; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.options.MemorySize; +import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.Table; +import org.apache.paimon.table.source.DataSplit; +import org.apache.paimon.table.source.IncrementalSplit; +import org.apache.paimon.table.source.Split; +import org.apache.paimon.table.source.Splits; +import org.apache.paimon.types.ArrayType; +import org.apache.paimon.types.DataType; +import org.apache.paimon.types.MapType; +import org.apache.paimon.types.MultisetType; +import org.apache.paimon.types.RowType; +import org.apache.paimon.types.VariantType; + +import java.util.HashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.function.LongFunction; + +/** + * The JVM heap BE's JNI reader of a paimon split holds, declared to BE's JNI heap gate + * ({@code TFileRangeDesc.jni_heap_bytes}) for a statement that sets {@code enable_jni_heap_admission}. + * + * <p>A JNI read of a {@link DataSplit} whose files overlap - a primary-key table written to since it was last + * compacted - merges them: paimon opens one file of every sorted run of a section at once, and an open + * parquet file keeps the compressed column chunks of its current row group in the heap until it moves + * on to the next, when the old and the new are there together for a moment. ORC keeps a stripe the + * same way. Sixteen such splits read at once are what runs a 2 GB heap out. So a split that has to be + * merged holds, steadily, the sum over its files of one row group - two for a file that has more than one + * - plus a dictionary page for every column of every file, a file storing a column for every primitive + * value a field nests. That takes all the files as one section, which is exact for the uncompacted tables + * that need the gate and over the mark for files that do not overlap, which are read one section after + * another. A split that needs no merging reads its files one after another and holds one file's share - + * unless a file of a primary-key table in it was written before paimon counted deleted rows: paimon plans + * such files as needing no merging, but merges them when it reads them, as they may hold deletes. + * + * <p>A diff read between two snapshots ({@code @incr} with {@code incrementalBetweenScanMode} diff) plans an + * {@link IncrementalSplit} instead, of the files only the earlier snapshot has and those only the later one + * has. Of a primary-key table it merges each of the two and compares the merges, which are open together, + * so it holds what one merge of all those files holds; of an append table it reads the later snapshot's + * files one after another. The other splits that are not DataSplits, those of the metadata system tables + * and of format tables, hold little and declare nothing. + * + * <p>A split is looked at through the wrappers paimon plans it in, which its reader reads through: the + * {@code QueryAuthSplit} every split of a table with query-auth rules (a catalog's row filter or column + * mask) comes in, and the {@code FallbackSplit} of a fallback-branch table, which wraps that in turn. + * + * <p>A file's row groups, dictionary pages and columns are those of the schema version it was written under + * ({@link DataFileMeta#schemaId()}): paimon's writers take the row group size and the largest parquet + * dictionary page ({@code parquet.dictionary.page.size}) from the table's options, and an ALTER TABLE that + * changes them, or the columns, leaves the files written before it as they are. + * A write job can also pass the row group size as a dynamic option, which no schema version records; a + * file written so is taken by its schema version's options, and declared below what it holds when the + * job's row groups were larger. + * + * <p>That steady hold is what is declared, not everything a merge ever holds: while it decodes, its column + * batches add brief peaks on top, measured at about three times the declaration for a split of a 126 MB + * file read on its own. The peaks are left to the heap outside the budget - the other half of it with + * jni_scanner_heap_budget_ratio at its default - which has room for some of the admitted readers to peak + * at once but not for all of them; a read that needs more fails as it does with enable_jni_heap_admission + * off. + * + * <p>Every column is taken as read. A narrow projection reads less, but a merge also reads every key + * column, the sequence number and the row kind whatever is projected, and declaring less than a merge + * holds steadily would let the gate admit more readers than its budget fits. + */ +final class PaimonJniHeapEstimate { + + // What paimon's writers fall back to when neither file.block-size nor the format's own option is + // set (CoreOptions.FILE_BLOCK_SIZE). + static final long DEFAULT_PARQUET_ROW_GROUP_BYTES = 128L * 1024 * 1024; + static final long DEFAULT_ORC_STRIPE_BYTES = 64L * 1024 * 1024; + // The largest dictionary page paimon's parquet writer keeps for a column unless the table's + // parquet.dictionary.page.size says otherwise (parquet's DEFAULT_DICTIONARY_PAGE_SIZE); an ORC file is + // allowed as much for each of its columns. + static final long DEFAULT_DICTIONARY_PAGE_BYTES = 1024L * 1024; + + // A primary-key table's diff read merges its files, as does its read of files that do not count their + // deleted rows; an append table's reads them one after another. + private final boolean primaryKeyTable; + // How the files written under a schema version are laid out, by its id, each worked out once. + private final LongFunction<FileLayout> layoutOfSchema; + private final Map<Long, FileLayout> layouts = new HashMap<>(); + + PaimonJniHeapEstimate(boolean primaryKeyTable, LongFunction<FileLayout> layoutOfSchema) { + this.primaryKeyTable = primaryKeyTable; + this.layoutOfSchema = layoutOfSchema; + } + + /** + * For the splits of {@code table}, each file laid out by the schema version it was written under: + * the table's own is at hand, and {@code schemaAt} reads the others by id. + */ + static PaimonJniHeapEstimate of(FileStoreTable table, LongFunction<TableSchema> schemaAt) { + TableSchema own = table.schema(); + return new PaimonJniHeapEstimate(!own.primaryKeys().isEmpty(), + schemaId -> FileLayout.of(schemaId == own.id() ? own : schemaAt.apply(schemaId))); + } + + /** For a table whose schema versions are not at hand: every file laid out by the options it shows. */ + static PaimonJniHeapEstimate of(Table table) { + FileLayout layout = FileLayout.of(table.options(), FileLayout.columns(table.rowType()), + table.primaryKeys().size()); + return new PaimonJniHeapEstimate(!table.primaryKeys().isEmpty(), schemaId -> layout); + } + + long bytesOf(Split split) { + // A fallback-branch table's FallbackDataSplit is a DataSplit itself. + Split read = Splits.underlying(split); + if (read instanceof DataSplit) { + DataSplit dataSplit = (DataSplit) read; + return readsFileByFile(dataSplit) ? largest(dataSplit.dataFiles()) : total(dataSplit.dataFiles()); + } + if (!(read instanceof IncrementalSplit)) { + return 0; Review Comment: [P2] Declare the chain files read through audit-log `ChainSplit` also reaches this zero-return branch. The ordinary-chain rejection in `resolveScanTable()` only applies to non-system handles: `$audit_log` over a chain table whose main branch lacks a partition can plan `FallbackSplitImpl(ChainSplit)` containing snapshot and delta files. `Splits.underlying()` removes the outer wrapper but leaves this split kind. The SDK's [ChainGroupReadTable.Read](https://github.com/apache/paimon/blob/release-2.1.0/paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java) explicitly accepts it and delegates to `MergeFileSplitRead.createChainReader()`, which merges those files while retaining their row groups and dictionaries. Returning 0 omits `jni_heap_bytes`, so these readers bypass admission entirely even when enabled. Estimate the chain files using their recorded physical branches and schemas, and cover an audit-log snapshot/delta scan. This is a separate split kind from the QueryAuthSplit and IncrementalSplit cases already fix ed; this finding does not assume chain-binlog support. ########## fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java: ########## @@ -2219,6 +2239,47 @@ && containsVariant(field.type()))) { return physicalVariantSchemaIds; } + /** + * What the JNI splits of {@code table} declare (PaimonJniHeapEstimate). A file holds row groups of the + * size set by the schema version it was written under, which an ALTER TABLE since leaves as it is, so + * the estimate reads the versions its files were written under, of the branch that planned its split: a + * scan.fallback-branch or scan.primary-branch table plans the splits of two branches, each numbering its + * schema versions on its own. $ro, $audit_log, $binlog and $row_tracking read their base table's files, + * under that table's schema ids, while showing its current options; the metadata system tables are + * estimated by the options they show. + */ + private ToLongFunction<Split> jniHeapBytes(Table table, PaimonTableHandle handle) { + // The base table of a wrapper, resolved as the timestamp-history checks resolve it. + Table physicalSchemaTable = withSchemaHistoryAuthentication(() -> resolveTimestampHistoryTable(table, handle)); + if (!(physicalSchemaTable instanceof FileStoreTable)) { + return PaimonJniHeapEstimate.of(table)::bytesOf; + } + FileStoreTable fileStoreTable = (FileStoreTable) physicalSchemaTable; + Map<FileStoreTable, PaimonJniHeapEstimate> branchEstimates = new java.util.IdentityHashMap<>(); + return split -> branchEstimates.computeIfAbsent(splitBranch(fileStoreTable, split), this::jniHeapEstimate) Review Comment: [P2] Preserve forced-merge sizing for audit-log readers `jniHeapBytes()` reduces the wrapper to its physical branch, losing `$audit_log`/`$binlog`'s read mode. A modern `merge-engine=first-row` table can pack a `rawConvertible` split with L2 keys `{1,100}` and L1 keys `{2,99}`, both with `deleteRowCount=0`. [AuditLogRead](https://github.com/apache/paimon/blob/release-2.1.0/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java) calls `forceKeepDelete()`, so the [raw-read provider](https://github.com/apache/paimon/blob/release-2.1.0/paimon-core/src/main/java/org/apache/paimon/table/source/splitread/PrimaryKeyTableRawFileSplitReadProvider.java) rejects this split and the SDK merges both sorted runs together. The estimator instead selects `largest(...)`, omitting the other file's concurrently retained row group and dictionaries. Concurrent readers can therefore exceed the intended steady-heap budget with admission enabled. Preserve the wrapper's forced-merge mode and sum these files; cover modern mixed-level audit/bin log splits. This is distinct from existing thread 4308681695's ordinary legacy files with absent delete counts, which the new predicate fixes. -- 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]
