morningman commented on code in PR #68713: URL: https://github.com/apache/doris/pull/68713#discussion_r4396309312
########## 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: Not changing the estimate here: the reader this describes is never created. **Confirmed as far as it goes.** `$audit_log` over a chain table plans `FallbackSplitImpl(ChainSplit)` for the partitions its main branch lacks (`AuditLogTable#newScan` hands the scan to the pair's `FallbackReadScan`), `Splits.underlying()` leaves the `ChainSplit`, and the estimate declares 0 for it. **Why no reader bypasses admission.** The range never reaches BE: 1. A chain table is always partitioned (`SchemaValidation`: "Chain table must be partition table."). 2. `$audit_log` reports the base table's partition keys (`AuditLogTable#partitionKeys`), so the scan node declares `PATH_PARTITION_KEYS`. 3. A split that is not a `DataSplit` goes to JNI with an empty partition-value map and no path (`buildJniScanRange`). `PaimonScanRange` keeps `isPartitionBearing()` false, so `PluginDrivenSplit` hands the engine null partition values and the path `connector://virtual`. 4. `FileQueryScanNode#splitToScanRange` then parses partition values from that path and fails with `Fail to parse columnsFromPath, expected: [dt], filePath: connector://virtual` while FE builds the scan ranges. That is the error #68897 recorded for a chain table's own data scan. So such a statement fails with the variable on or off, before any reader opens. `$binlog` would not get further anyway: `BinlogRead#createReader` casts each split to `DataSplit`. **Where it belongs.** Doris does not read chain tables yet (#68897). Declaring a `ChainSplit` (each file's branch from `fileBranchMapping()`, laid out by that branch's schema versions) should come with chain support, once such a split can be read at all; 7e152488295 says so in the class comment. Separately, these system tables should fail with #68897's clear message instead of the columnsFromPath one. That goes in a PR of its own, which I'll link here. -- 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]
