924060929 commented on code in PR #69298: URL: https://github.com/apache/doris/pull/69298#discussion_r4323380291
########## 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: Reproduced with real InMemoryCatalog manifests: after INT-to-BIGINT promotion, warming only the data or delete manifest, removing the partition source, and expiring its schemas, all four identity/truncate and cache-order cases returned zero matching equality deletes instead of one. The fix stays in the common scan-local partition adapter: UNKNOWN targets normalize Integer/Long to Long and Float/Double to Double, while preserving the actual value and leaving shared cached files untouched. Decimal precision promotion already has the same Java type and scale. Added concurrent original-schema/current-UNKNOWN projections, a different-partition negative check, and SDK partition-wrapper equality/hash tests including non-exact floats, NaN, signed zero and large integers. Validation: all 1,526 Iceberg tests and 56 FE lifecycle tests completed with zero failures/errors; one existing live-connectivity test is skipped. Full FE build and Checkstyle validation are recorded with this update. ########## 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: Updated the pre-capture assertion to CancellationException, matching QueryTaskGroups.get() and the terminal-cancellation contract. No production exception behavior is changed. All four CANCELLED/TIMEOUT and pre-/post-capture parameterizations pass, and the full StmtExecutorTest class (54 tests) plus QueryTaskGroupsTest (2 tests) pass. -- 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]
