github-actions[bot] commented on code in PR #67810: URL: https://github.com/apache/doris/pull/67810#discussion_r3981938742
########## fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergPartitionStatsTable.java: ########## @@ -0,0 +1,307 @@ +// 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.datasource.iceberg.action; + +import org.apache.doris.common.security.authentication.ExecutionAuthenticator; + +import org.apache.iceberg.BaseTable; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.HasTableOperations; +import org.apache.iceberg.ManifestFile; +import org.apache.iceberg.ManifestListFile; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.Table; +import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.io.PositionOutputStream; +import org.apache.iceberg.io.SeekableInputStream; +import org.apache.iceberg.metrics.LoggingMetricsReporter; + +import java.io.IOException; +import java.io.UncheckedIOException; +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.Callable; +import java.util.concurrent.atomic.AtomicBoolean; + +/** A borrowed, local table view for the SDK's shared partition-statistics worker pool. */ +final class IcebergPartitionStatsTable extends BaseTable { + private final Table delegate; + private final ExecutionAuthenticator authenticator; + private final FileIO authenticatedIo; + + IcebergPartitionStatsTable(Table table, ExecutionAuthenticator authenticator) { + super(((HasTableOperations) table).operations(), table.name(), + table instanceof BaseTable ? ((BaseTable) table).reporter() : LoggingMetricsReporter.instance()); + this.delegate = table; + this.authenticator = Objects.requireNonNull(authenticator, "authenticator is null"); + this.authenticatedIo = new AuthenticatedFileIO(table.io()); + } + + @Override + public FileIO io() { + return authenticatedIo; + } + + @Override + public Map<Integer, PartitionSpec> specs() { + return unchecked(delegate::specs); + } + + @Override + public Snapshot snapshot(long snapshotId) { Review Comment: [P1] Avoid one authentication transition per manifest entry Iceberg 1.11 calls `table.snapshot(entry.snapshotId())` inside its loop over every manifest entry. This override turns that in-memory metadata lookup into `authenticator.execute` -> Hadoop `doAs`/`getUGI`; the Kerberos UGI lookup is synchronized. Large tables can therefore perform millions of needless authentication transitions and serialize worker progress even though `BaseTable.snapshot` only reads captured `TableMetadata`. Keep authentication around the actual `FileIO` and lazy stream operations, but serve snapshot metadata without re-entering the authenticator, and add an invocation-count regression test. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergComputePartitionStatsAction.java: ########## @@ -0,0 +1,116 @@ +// 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.datasource.iceberg.action; + +import org.apache.doris.catalog.Column; +import org.apache.doris.catalog.Env; +import org.apache.doris.catalog.TableIf; +import org.apache.doris.catalog.Type; +import org.apache.doris.common.ArgumentParsers; +import org.apache.doris.common.UserException; +import org.apache.doris.common.security.authentication.ExecutionAuthenticator; +import org.apache.doris.datasource.iceberg.IcebergExternalTable; +import org.apache.doris.datasource.iceberg.IcebergMetadataOps; +import org.apache.doris.info.PartitionNamesInfo; +import org.apache.doris.nereids.trees.expressions.Expression; + +import org.apache.iceberg.PartitionStatisticsFile; +import org.apache.iceberg.PartitionStatsHandler; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.Table; + +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Optional; + +/** Computes and registers Iceberg partition statistics for a selected snapshot. */ +public class IcebergComputePartitionStatsAction extends BaseIcebergAction { + private static final String SNAPSHOT_ID = "snapshot_id"; + private final ExecutionAuthenticator authenticator; + + public IcebergComputePartitionStatsAction(Map<String, String> properties, + Optional<PartitionNamesInfo> partitionNamesInfo, Optional<Expression> whereCondition, + IcebergMetadataOps metadataOps) { + super(IcebergExecuteActionFactory.COMPUTE_PARTITION_STATS, properties, partitionNamesInfo, + whereCondition, metadataOps); + this.authenticator = metadataOps.getExecutionAuthenticator(); + } + + @Override + protected void registerIcebergArguments() { + namedArguments.registerOptionalArgument(SNAPSHOT_ID, + "Snapshot ID to compute partition statistics for (defaults to the current snapshot)", + null, ArgumentParsers.longRange(SNAPSHOT_ID, Long.MIN_VALUE, Long.MAX_VALUE)); + } + + @Override + protected void validateIcebergAction() throws UserException { + validateNoPartitions(); + validateNoWhereCondition(); + } + + @Override + protected List<Column> getResultSchema() { + return Collections.singletonList(new Column("partition_statistics_file", Type.STRING, true, + "Path of the partition statistics file")); + } + + @Override + protected List<List<String>> executeAction(TableIf table) throws UserException { + IcebergExternalTable dorisTable = (IcebergExternalTable) table; + // Let the command retry a catalog-generation fence before any statistics file is written. + Table icebergTable = getWritableIcebergTable(table); + try { + Long requestedSnapshotId = namedArguments.getLong(SNAPSHOT_ID); + Snapshot snapshot; + if (requestedSnapshotId != null) { + snapshot = icebergTable.snapshot(requestedSnapshotId); + if (snapshot == null) { + throw new UserException("Snapshot not found: " + requestedSnapshotId); + } + } else { + snapshot = icebergTable.currentSnapshot(); + if (snapshot == null) { + return Collections.emptyList(); + } + } + + // Keep the selected snapshot even if the table head changes during computation. + long snapshotId = snapshot.snapshotId(); + Table statisticsTable = new IcebergPartitionStatsTable(icebergTable, authenticator); + PartitionStatisticsFile file = PartitionStatsHandler.computeAndWriteStatsFile(statisticsTable, snapshotId); Review Comment: [P1] Support ORC tables before exposing this action Iceberg 1.11 chooses the statistics-file format from the supplied table's `write.format.default`, but its internal writer has providers only for Avro and Parquet. The new integration test demonstrates the consequence: reuse of an older file works, then the first new snapshot on an ORC-default table fails with `unregistered internal data format: ORC`. Doris already supports ORC Iceberg tables and the factory exposes this procedure without a format restriction, so please decouple the statistics-file format (for example, present Parquet through the scoped table view) or gate the action consistently, and replace the expected failure with successful ORC compute/readback coverage. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergComputePartitionStatsAction.java: ########## @@ -0,0 +1,116 @@ +// 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.datasource.iceberg.action; + +import org.apache.doris.catalog.Column; +import org.apache.doris.catalog.Env; +import org.apache.doris.catalog.TableIf; +import org.apache.doris.catalog.Type; +import org.apache.doris.common.ArgumentParsers; +import org.apache.doris.common.UserException; +import org.apache.doris.common.security.authentication.ExecutionAuthenticator; +import org.apache.doris.datasource.iceberg.IcebergExternalTable; +import org.apache.doris.datasource.iceberg.IcebergMetadataOps; +import org.apache.doris.info.PartitionNamesInfo; +import org.apache.doris.nereids.trees.expressions.Expression; + +import org.apache.iceberg.PartitionStatisticsFile; +import org.apache.iceberg.PartitionStatsHandler; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.Table; + +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Optional; + +/** Computes and registers Iceberg partition statistics for a selected snapshot. */ +public class IcebergComputePartitionStatsAction extends BaseIcebergAction { + private static final String SNAPSHOT_ID = "snapshot_id"; + private final ExecutionAuthenticator authenticator; + + public IcebergComputePartitionStatsAction(Map<String, String> properties, + Optional<PartitionNamesInfo> partitionNamesInfo, Optional<Expression> whereCondition, + IcebergMetadataOps metadataOps) { + super(IcebergExecuteActionFactory.COMPUTE_PARTITION_STATS, properties, partitionNamesInfo, + whereCondition, metadataOps); + this.authenticator = metadataOps.getExecutionAuthenticator(); + } + + @Override + protected void registerIcebergArguments() { + namedArguments.registerOptionalArgument(SNAPSHOT_ID, + "Snapshot ID to compute partition statistics for (defaults to the current snapshot)", + null, ArgumentParsers.longRange(SNAPSHOT_ID, Long.MIN_VALUE, Long.MAX_VALUE)); + } + + @Override + protected void validateIcebergAction() throws UserException { + validateNoPartitions(); + validateNoWhereCondition(); + } + + @Override + protected List<Column> getResultSchema() { + return Collections.singletonList(new Column("partition_statistics_file", Type.STRING, true, + "Path of the partition statistics file")); + } + + @Override + protected List<List<String>> executeAction(TableIf table) throws UserException { + IcebergExternalTable dorisTable = (IcebergExternalTable) table; + // Let the command retry a catalog-generation fence before any statistics file is written. + Table icebergTable = getWritableIcebergTable(table); + try { + Long requestedSnapshotId = namedArguments.getLong(SNAPSHOT_ID); + Snapshot snapshot; + if (requestedSnapshotId != null) { + snapshot = icebergTable.snapshot(requestedSnapshotId); + if (snapshot == null) { + throw new UserException("Snapshot not found: " + requestedSnapshotId); + } + } else { + snapshot = icebergTable.currentSnapshot(); + if (snapshot == null) { + return Collections.emptyList(); + } + } + + // Keep the selected snapshot even if the table head changes during computation. + long snapshotId = snapshot.snapshotId(); + Table statisticsTable = new IcebergPartitionStatsTable(icebergTable, authenticator); + PartitionStatisticsFile file = PartitionStatsHandler.computeAndWriteStatsFile(statisticsTable, snapshotId); + if (file == null) { + return Collections.emptyList(); + } + + // Follow the SDK commit contract, including reuse of an existing statistics file. + icebergTable.updatePartitionStatistics().setPartitionStatistics(file).commit(); + Env.getCurrentEnv().getExtMetaCacheMgr().invalidateTableCache(dorisTable); + return Collections.singletonList(Collections.singletonList(file.path())); + } catch (Exception e) { Review Comment: [P1] Refresh metadata after an unknown commit outcome A `CommitStateUnknownException` can be raised after the metadata update is durable; the new integration test simulates exactly that and observes the registered statistics entry. This catch path leaves the local invalidation unexecuted and also prevents `ExecuteActionCommand` from publishing its follower refresh log, so every FE can keep serving pre-commit metadata after a real commit. Do not retry the mutation or delete the file, but conservatively invalidate and propagate the refresh before rethrowing, and make the command-level test require those cache/edit-log interactions. ########## fe/fe-core/src/test/java/org/apache/doris/qe/StmtExecutorTest.java: ########## @@ -312,6 +315,35 @@ public Void answer(InvocationOnMock invocation) { executor.sendTextResultRow(resultSet); } + @Test + public void testProxyResultSetRetainsMetadataForZeroAndMultipleRows() throws IOException { + for (int rowCount : new int[] {0, 1, 2}) { + ConnectContext ctx = Mockito.mock(ConnectContext.class); + ProxyMysqlChannel channel = new ProxyMysqlChannel(); + Mockito.when(ctx.getConnectType()).thenReturn(ConnectType.MYSQL); Review Comment: [P1] Support Arrow Flight clients on follower FEs `ExecuteActionCommand` is `ForwardWithSync`, but this new result path is tested only with a MySQL proxy channel. On a non-master FE, an Arrow Flight connection fails while building the forward request because `buildStmtForwardParams` unconditionally calls `ctx.getMysqlChannel().clientDeprecatedEOF()`, and `FlightSqlConnectContext.getMysqlChannel()` throws. Merely guarding that read is not enough: the request carries no result protocol and the master returns MySQL packet buffers rather than a Flight `ResultSet`. The action therefore works or fails according to which FE receives the same Flight request. Please make forwarding/result transport protocol-aware (or explicitly route/reject this command) and cover empty, one-row, and multi-row results through a follower Flight connection. ########## fe/fe-core/src/test/java/org/apache/doris/qe/StmtExecutorTest.java: ########## @@ -312,6 +315,35 @@ public Void answer(InvocationOnMock invocation) { executor.sendTextResultRow(resultSet); } + @Test + public void testProxyResultSetRetainsMetadataForZeroAndMultipleRows() throws IOException { + for (int rowCount : new int[] {0, 1, 2}) { + ConnectContext ctx = Mockito.mock(ConnectContext.class); + ProxyMysqlChannel channel = new ProxyMysqlChannel(); + Mockito.when(ctx.getConnectType()).thenReturn(ConnectType.MYSQL); + Mockito.when(ctx.getMysqlChannel()).thenReturn(channel); + Mockito.when(ctx.getSessionVariable()).thenReturn(VariableMgr.newSessionVariable()); + QueryState state = new QueryState(); + Mockito.when(ctx.getState()).thenReturn(state); + List<List<String>> rows = Lists.newArrayList(); + for (int index = 0; index < rowCount; index++) { + rows.add(Collections.singletonList("stats-" + index + ".parquet")); + } + ResultSet result = new CommonResultSet(new CommonResultSetMetaData(Collections.singletonList( + new Column("partition_statistics_file", PrimitiveType.STRING, true))), rows); + new StmtExecutor(ctx, new OriginStatement("", 0), true).sendResultSet(result); Review Comment: [P1] Preserve binary mode for zero-parameter prepared executes The common `compute_partition_stats()` call has no placeholders. For a server-prepared execution, `handleExecute` saves `prepareExecuteBuffer` only when `paramCount > 0`, so the follower omits it from `TMasterOpRequest`; the master then takes `queryRetry()` through the text-protocol proxy constructor and serializes result rows with `sendTextResultRow()`. The client is waiting for binary rows, whose first byte/null bitmap differ. This test uses that same text-only constructor and checks only packet count/metadata, so it cannot catch the corruption. Forward an explicit COM_STMT_EXECUTE/binary-result flag independently of parameter bytes, and assert binary row framing for a zero-argument action on a follower. ########## fe/fe-core/src/test/java/org/apache/doris/qe/StmtExecutorTest.java: ########## @@ -312,6 +315,35 @@ public Void answer(InvocationOnMock invocation) { executor.sendTextResultRow(resultSet); } + @Test + public void testProxyResultSetRetainsMetadataForZeroAndMultipleRows() throws IOException { + for (int rowCount : new int[] {0, 1, 2}) { + ConnectContext ctx = Mockito.mock(ConnectContext.class); + ProxyMysqlChannel channel = new ProxyMysqlChannel(); + Mockito.when(ctx.getConnectType()).thenReturn(ConnectType.MYSQL); + Mockito.when(ctx.getMysqlChannel()).thenReturn(channel); + Mockito.when(ctx.getSessionVariable()).thenReturn(VariableMgr.newSessionVariable()); + QueryState state = new QueryState(); + Mockito.when(ctx.getState()).thenReturn(state); + List<List<String>> rows = Lists.newArrayList(); + for (int index = 0; index < rowCount; index++) { + rows.add(Collections.singletonList("stats-" + index + ".parquet")); + } + ResultSet result = new CommonResultSet(new CommonResultSetMetaData(Collections.singletonList( + new Column("partition_statistics_file", PrimitiveType.STRING, true))), rows); + new StmtExecutor(ctx, new OriginStatement("", 0), true).sendResultSet(result); + List<ByteBuffer> packets = channel.getProxyResultBufferList(); + Assertions.assertEquals(3 + rowCount, packets.size()); + Assertions.assertEquals(1, packets.get(0).get(0)); + ByteBuffer definition = packets.get(1).duplicate(); + byte[] bytes = new byte[definition.remaining()]; + definition.get(bytes); + Assertions.assertTrue(new String(bytes, StandardCharsets.UTF_8).contains("partition_statistics_file")); + Assertions.assertEquals(QueryState.MysqlStateType.EOF, state.getStateType()); Review Comment: [P1] Exercise and fix the complete follower path This assertion captures the state that breaks forwarding: `sendResultSet` leaves a successful result at `EOF`, while `ConnectProcessor.proxyExecute` assigns status 0 and `affectedRows` only for `OK`. A follower still replays these packet buffers, so this test passes, but it receives `ERR_UNKNOWN_ERROR`, loses returned-row accounting, records an audit error, and stops any remaining multi-statements. Please cover the complete master-to-follower path for 0/1/multiple rows, treat a successful result-set EOF as proxy success, and carry the real returned-row count. -- 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]
