github-actions[bot] commented on code in PR #69298: URL: https://github.com/apache/doris/pull/69298#discussion_r4322080662
########## fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergPartitionValueAdapter.java: ########## @@ -0,0 +1,297 @@ +// 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.iceberg; + +import org.apache.iceberg.ContentFile; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.PartitionData; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.Types; + +import java.math.BigDecimal; +import java.nio.ByteBuffer; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** Query-local partition adaptation; all other content-file metadata is delegated unchanged. */ +final class IcebergPartitionValueAdapter { + private final int[] sourcePositions; + private final Type[] sourceTypes; + private final Type[] targetTypes; + private final PartitionData template; + + IcebergPartitionValueAdapter(Types.StructType source, Types.StructType target) { + int size = target.fields().size(); + sourcePositions = new int[size]; + sourceTypes = new Type[size]; + targetTypes = new Type[size]; + template = new PartitionData(target); + for (int i = 0; i < size; i++) { + Types.NestedField field = target.fields().get(i); + targetTypes[i] = field.type(); + sourcePositions[i] = -1; + for (int j = 0; j < source.fields().size(); j++) { + if (source.fields().get(j).fieldId() == field.fieldId()) { + sourcePositions[i] = j; + sourceTypes[i] = source.fields().get(j).type(); + break; + } + } + if (sourcePositions[i] < 0) { + throw new IllegalArgumentException("Missing cached partition field " + field.fieldId()); + } + } + } + + DataFile adapt(DataFile file) { + return new DataView(file, adapt((PartitionData) file.partition())); + } + + DeleteFile adapt(DeleteFile file) { + return new DeleteView(file, adapt((PartitionData) file.partition())); + } + + PartitionData adapt(PartitionData source) { + // Copies reuse the Avro schema; do not reconstruct/serialize it for every file. + PartitionData original = source.copy(); + PartitionData result = template.copy(); + for (int i = 0; i < sourcePositions.length; i++) { + if (sourcePositions[i] >= 0) { + result.set(i, convert(original.get(sourcePositions[i], Object.class), sourceTypes[i], targetTypes[i])); + } + } + return result; + } + + private static Object convert(Object value, Type source, Type target) { + // Dropping a source leaves UNKNOWN in the current spec. Iceberg still exposes the physical + // value from the manifest; preserve it in the local projection rather than replacing it with null. + if (value == null || source.equals(target) || target.typeId() == Type.TypeID.UNKNOWN) { Review Comment: [P1] Canonicalize unknown partition values before delete matching With manifest caching enabled, create v2 data manifest D and a later equality-delete manifest E under identity(p INT), then promote p to BIGINT and scan a snapshot before E: D is cached with Long(7), while E stays cold. Remove the p partition, drop p, append under the new spec, expire snapshots with cleanExpiredMetadata(true), and REFRESH TABLE. The retained cold E now recovers Integer(7) from its original header, whereas D keeps its promoted cached value. Both forScan calls target UNKNOWN, so this branch preserves incompatible representations. Iceberg's [partition-key comparison](https://github.com/apache/iceberg/blob/apache-iceberg-1.12.0/core/src/main/java/org/apache/iceberg/util/StructLikeWrapper.java#L85-L90) turns their comparison failure into inequality; DeleteFileIndex omits E, so deleted rows become visible without any error or SDK fallback. This differs from the existing cold schema-recovery thread: it requires a promoted warm payload and fails during delete matching her e. Canonicalize data/delete partition values consistently when the source is UNKNOWN, and cover promotion plus mixed cache warmth and source-schema cleanup together. ########## fe/fe-core/src/test/java/org/apache/doris/qe/StmtExecutorTest.java: ########## @@ -134,6 +140,111 @@ public void testCommittedTsoErrorSurvivesPlannerWrapping() throws Exception { } } + @ParameterizedTest + @ValueSource(booleans = {false, true}) + @ResourceLock("global") + public void testQueryCallbacksWithoutCoordinator(boolean planningFailure) throws Exception { + ConnectContext context = Mockito.spy(connectContext); + context.getState().reset(); + Mockito.doReturn(Mockito.mock(ResultSender.class)).when(context).getResultSender(); + TUniqueId queryId = new TUniqueId(0x22050L, planningFailure ? 1L : 2L); + AtomicInteger cleanup = new AtomicInteger(); + QeProcessorImpl.INSTANCE.registerQueryFinishCallback(DebugUtil.printId(queryId), cleanup::incrementAndGet); + try { + StmtExecutor executor = new StmtExecutor(context, "explain select 1"); + if (planningFailure) { + try (MockedConstruction<NereidsPlanner> planners = Mockito.mockConstruction(NereidsPlanner.class, + (planner, construction) -> Mockito.doThrow(new NereidsException("planning failed", + new UserException("planning failed"))) + .when(planner).plan(Mockito.any(StatementBase.class), Mockito.any(TQueryOptions.class)))) { + executor.execute(queryId); + Assertions.assertEquals(1, planners.constructed().size()); + } + Assertions.assertEquals(QueryState.MysqlStateType.ERR, context.getState().getStateType()); + } else { + executor.execute(queryId); + Assertions.assertNotEquals(QueryState.MysqlStateType.ERR, context.getState().getStateType(), + context.getState().getErrorMessage()); + } + Assertions.assertNull(QeProcessorImpl.INSTANCE.getCoordinator(queryId)); + Assertions.assertEquals(1, cleanup.get(), "planning-only attempts must release registered callbacks"); + QeProcessorImpl.INSTANCE.unregisterQuery(queryId); + Assertions.assertEquals(1, cleanup.get(), "later finalization must not repeat cleanup"); + } finally { + QeProcessorImpl.INSTANCE.unregisterQuery(queryId); + } + } + + @ParameterizedTest + @CsvSource({"false, CANCELLED", "true, CANCELLED", "false, TIMEOUT", "true, TIMEOUT"}) + @ResourceLock("global") + public void testPlanningCancellationClosesCapturedTaskGroups(boolean cancelBeforeCapture, + TStatusCode statusCode) throws Exception { + ConnectContext context = Mockito.spy(connectContext); + context.setStatementContext(new StatementContext(context, new OriginStatement("select 1", 0))); + context.setQueryId(new TUniqueId(0x22051L, statusCode.getValue() * 2L + (cancelBeforeCapture ? 1 : 0))); + StmtExecutor executor = new StmtExecutor(context, "select 1"); + context.setExecutor(executor); + Status reason = new Status(statusCode, "cancel before coordinator publication"); + if (cancelBeforeCapture) { + executor.cancel(reason, false); + } + ConnectorSession session = ConnectorSessionBuilder.from(context).withCatalogId(1).build(); + try { + if (cancelBeforeCapture) { + Assertions.assertThrows(java.util.concurrent.RejectedExecutionException.class, Review Comment: [P2] Update the pre-capture cancellation assertion Both `cancelBeforeCapture=true` cases now fail this assertion. The earlier `executor.cancel(...)` makes `ConnectorSessionBuilder.build()` immediately close the newly registered `QueryTaskGroups`; `getTaskScheduler(SPLIT)` then reaches `QueryTaskGroups.get()`, which now throws `CancellationException`. That is not a `RejectedExecutionException`, so both the CANCELLED and TIMEOUT parameterizations fail instead of checking the intended terminal cancellation. Expect `CancellationException` here, as the new `QueryTaskGroupsTest` already does. -- 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]
