github-actions[bot] commented on code in PR #67395: URL: https://github.com/apache/doris/pull/67395#discussion_r4132269204
########## fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/DorisIOManager.java: ########## @@ -0,0 +1,392 @@ +// 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.paimon; + +import org.apache.paimon.disk.BufferFileReader; +import org.apache.paimon.disk.BufferFileWriter; +import org.apache.paimon.disk.FileIOChannel; +import org.apache.paimon.disk.IOManager; +import org.apache.paimon.memory.Buffer; + +import java.io.File; +import java.io.IOException; +import java.io.UncheckedIOException; +import java.nio.channels.FileChannel; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +/** Paimon IOManager adapter which charges temporary I/O to Doris spill management. */ +final class DorisIOManager implements IOManager { + interface SpillAccountant { + String[] getSpillDirectories() throws IOException; + + void reserve(String path, long bytes) throws IOException; + + void rollback(String path, long bytes); + + void commitWrite(String path, long bytes); + + void recordRead(String path, long bytes); + + void release(String path, long bytes); + } + + static final class SpillDirectoryCleanupException extends IOException { + private SpillDirectoryCleanupException(Exception cause) { + super("Failed to eagerly clean a Paimon spill directory", cause); + } + } + + private final SpillAccountant accountant; + private final Map<String, Long> channelBytes = new ConcurrentHashMap<>(); + private final Map<String, Integer> activeChannelWriters = new ConcurrentHashMap<>(); + private volatile IOManager delegate; + + static DorisIOManager create(long nativeSpillSession) { + return new DorisIOManager(new NativeSpillAccountant(nativeSpillSession)); + } + + DorisIOManager(SpillAccountant accountant) { + this(null, accountant); + } + + DorisIOManager(IOManager delegate, SpillAccountant accountant) { + this.accountant = accountant; + this.delegate = delegate; + } + + private IOManager delegate() throws IOException { + if (delegate == null) { + synchronized (this) { + if (delegate == null) { + String[] spillDirectories = accountant.getSpillDirectories(); + if (spillDirectories == null || spillDirectories.length == 0) { + throw new IOException("Doris spill manager returned no available directories"); + } + delegate = IOManager.create(spillDirectories); + } + } + } + return delegate; + } + + private IOManager uncheckedDelegate() { + try { + return delegate(); + } catch (IOException e) { + throw new UncheckedIOException("Failed to initialize the Doris spill directory", e); + } + } + + @Override + public FileIOChannel.ID createChannel() { + return uncheckedDelegate().createChannel(); + } + + @Override + public FileIOChannel.ID createChannel(String prefix) { + return uncheckedDelegate().createChannel(prefix); Review Comment: [P1] Account for Paimon files written directly into spill paths — Paimon creates or downloads lookup SSTs through a file returned by `createChannel(prefix).getPathFile()` and writes clustering SSTs under `pickTempDir()`. These methods only return Doris-managed paths; the direct SDK file writes never call `createBufferFileWriter()` or `AccountingBufferFileWriter.writeBlock()`, where spill bytes are reserved and charged. Supported lookup or `pk-clustering-override` writes can therefore exceed Doris's spill budget and fill the BE disk while usage metrics omit those files; Paimon's lookup cache defaults to an unbounded size. Account and bound direct file growth in both paths, and test low-quota lookup and clustering writes. ########## fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorContext.java: ########## @@ -38,6 +38,11 @@ public interface ConnectorContext { /** Returns the catalog ID. */ long getCatalogId(); + /** Returns the persisted Doris cluster id used to namespace external commit identities. */ + default int getClusterId() { Review Comment: [P2] Record the new methods in the connector SPI baseline — This adds `ConnectorContext#getClusterId():int`, and the PR also adds `ConnectorWriteHandle#getStaticPartitionNullKeys():java.util.Set`, but `connector-plugin-surface.txt` records neither signature. Both interfaces are in `ConnectorPluginSurfaceTest.FROZEN_TYPES`; its `getMethods()` scan therefore adds both signatures to the actual surface and the exact baseline assertion fails whenever this test runs. Add both entries to the baseline; the SPI major version is already bumped to 11. ########## fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonVariantProjection.java: ########## @@ -0,0 +1,184 @@ +// 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.paimon; + +import org.apache.paimon.data.DataGetters; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.data.variant.GenericVariant; +import org.apache.paimon.data.variant.GenericVariantBuilder; +import org.apache.paimon.data.variant.Variant; +import org.apache.paimon.data.variant.VariantMetadataUtils; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowType; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +/** Bridges Doris Variant access paths to Paimon's Variant extraction RowType. */ +final class PaimonVariantProjection { + private static final String FIELD_NAME_PREFIX = "__doris_variant_field_"; + private static final int VARIANT_VALUE_INDEX = 0; + private static final int VARIANT_METADATA_INDEX = 1; + private static final int VARIANT_FIELD_COUNT = 2; + + private final RowType readType; + private final PathNode root; + + private PaimonVariantProjection(RowType readType, PathNode root) { + this.readType = readType; + this.root = root; + } + + static PaimonVariantProjection create(List<List<String>> paths, String timeZone) { + if (paths == null || paths.isEmpty()) { + return null; + } + List<DataField> fields = new ArrayList<>(paths.size()); + PathNode root = new PathNode(); + for (int fieldIndex = 0; fieldIndex < paths.size(); fieldIndex++) { + List<String> path = paths.get(fieldIndex); + if (!supportsObjectPath(path) || !root.add(path, fieldIndex)) { + return null; + } + fields.add(new DataField( + fieldIndex, + FIELD_NAME_PREFIX + fieldIndex, + DataTypes.VARIANT(), + VariantMetadataUtils.buildVariantMetadata(toPaimonPath(path), false, timeZone))); + } + return new PaimonVariantProjection(new RowType(fields), root); + } + + RowType readType() { + return readType; + } + + Variant materialize(DataGetters record, int fieldIndex) { + InternalRow extracted = record.getRow(fieldIndex, readType.getFieldCount()); + GenericVariantBuilder builder = new GenericVariantBuilder(false); + appendObject(builder, root, extracted); + return builder.result(); + } + + private static boolean supportsObjectPath(List<String> path) { + if (path == null || path.isEmpty()) { + return false; + } + for (String segment : path) { + if (segment == null || segment.isEmpty() || segment.indexOf('.') >= 0 + || segment.indexOf('[') >= 0 || segment.indexOf(';') >= 0 + || isIntegerSegment(segment)) { + return false; + } + } + return true; + } + + private static boolean isIntegerSegment(String segment) { + int offset = segment.startsWith("-") ? 1 : 0; + if (offset == segment.length()) { + return false; + } + for (int i = offset; i < segment.length(); i++) { + if (!Character.isDigit(segment.charAt(i))) { + return false; + } + } + return true; + } + + private static String toPaimonPath(List<String> path) { + return "$." + String.join(".", path); + } + + private static boolean hasValue(PathNode node, InternalRow extracted) { + if (node.fieldIndex >= 0) { + return hasExtractedVariant(extracted, node.fieldIndex); + } + for (PathNode child : node.children.values()) { + if (hasValue(child, extracted)) { + return true; + } + } + return false; + } + + private static boolean hasExtractedVariant(InternalRow extracted, int fieldIndex) { + InternalRow variant = extracted.getRow(fieldIndex, VARIANT_FIELD_COUNT); Review Comment: [P2] Check the projected field null bit before reading its Variant row — Paimon emits a null field when a requested object key is absent. For a primary-key scan with JNI IOManager enabled and enough overlapping readers to trigger merge spill, the reloaded value is an `OffsetRow` over `BinaryRow`; `getRow()` on that null field reads its zero slot as a nested row. This helper then inspects unrelated header bits and can throw `Paimon projected Variant must contain both value and metadata` for a valid missing key. Check `extracted.isNullAt(fieldIndex)` before `getRow()`, and cover a missing path with forced merge spill. -- 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]
