@Override
    public List<OptExpression> transform(OptExpression input, OptimizerContext context) {
        LogicalScanOperator scanOperator = (LogicalScanOperator) input.getOp();
        ColumnRefSet requiredOutputColumns = context.getTaskContext().getRequiredColumns();

        Set<ColumnRefOperator> scanColumns =
                scanOperator.getColRefToColumnMetaMap().keySet().stream().filter(requiredOutputColumns::contains)
                        .collect(Collectors.toSet());
        scanColumns.addAll(Utils.extractColumnRef(scanOperator.getPredicate()));

        checkPartitionColumnType(scanOperator, scanColumns, context);

        // make sure there is at least one materialized column in new output columns.
        // if not, we have to choose one materialized column from scan operator output columns
        // with the minimal cost.
        boolean canUseAnyColumn = false;

        // if this scan operator columns are all partitions columns(like iceberg table)
        // we have to take partition columns are materialized columns and read them from files.
        // And we can not use `canUseAnyColumn` optimization either.
        boolean allPartitionColumns =
                scanOperator.getPartitionColumns()
                        .containsAll(scanOperator.getColRefToColumnMetaMap().values().stream().map(x -> x.getName()).collect(
                                Collectors.toList()));

        if (!containsMaterializedColumn(scanOperator, scanColumns)) {
            List<ColumnRefOperator> preOutputColumns =
                    new ArrayList<>(scanOperator.getColRefToColumnMetaMap().keySet());
            List<ColumnRefOperator> outputColumns = preOutputColumns.stream()
                    .filter(column -> !column.getType().getPrimitiveType().equals(PrimitiveType.UNKNOWN_TYPE))
                    .collect(Collectors.toList());

            int smallestIndex = -1;
            int smallestColumnLength = Integer.MAX_VALUE;
            for (int index = 0; index < outputColumns.size(); ++index) {
                if (!allPartitionColumns && isPartitionColumn(scanOperator, outputColumns.get(index).getName())) {
                    continue;
                }

                if (smallestIndex == -1) {
                    smallestIndex = index;
                }
                Type columnType = outputColumns.get(index).getType();
                if (columnType.isScalarType() && columnType.isSupported()) {
                    int columnLength = columnType.getTypeSize();
                    if (columnLength < smallestColumnLength) {
                        smallestIndex = index;
                        smallestColumnLength = columnLength;
                    }
                }
            }
            Preconditions.checkArgument(smallestIndex != -1);
            scanColumns.add(outputColumns.get(smallestIndex));
            canUseAnyColumn = true;
        }

        if (allPartitionColumns || !context.getSessionVariable().isEnableCountStarOptimization()) {
            canUseAnyColumn = false;
        }

        if (scanOperator.getOutputColumns().equals(new ArrayList<>(scanColumns))) {
            scanOperator.getScanOptimzeOption().setCanUseAnyColumn(canUseAnyColumn);
            return Collections.emptyList();
        } else {
            try {
                Class<? extends LogicalScanOperator> classType = scanOperator.getClass();
                Map<ColumnRefOperator, Column> newColumnRefMap = scanColumns.stream()
                        .collect(Collectors.toMap(identity(), scanOperator.getColRefToColumnMetaMap()::get));
                LogicalScanOperator newScanOperator =
                        classType.getConstructor(Table.class, Map.class, Map.class, long.class,
                                ScalarOperator.class).newInstance(
                                scanOperator.getTable(),
                                newColumnRefMap,
                                scanOperator.getColumnMetaToColRefMap(),
                                scanOperator.getLimit(),
                                scanOperator.getPredicate());
                newScanOperator.getScanOptimzeOption().setCanUseAnyColumn(canUseAnyColumn);
                newScanOperator.setScanOperatorPredicates(scanOperator.getScanOperatorPredicates());
                newScanOperator.setTableVersionRange(scanOperator.getTableVersionRange());

                return Lists.newArrayList(new OptExpression(newScanOperator));
            } catch (Exception e) {
                throw new StarRocksPlannerException(e.getMessage(), ErrorType.INTERNAL_ERROR);
            }
        }
    }
@Test
    public void transformIcebergWithAllPartitionColumns(@Mocked IcebergTable table,
                                                        @Mocked OptimizerContext context,
                                                        @Mocked TaskContext taskContext) {
        OptExpression scan = new OptExpression(
                new LogicalIcebergScanOperator(table,
                        scanColumnMap, Maps.newHashMap(), -1, null, TableVersionRange.empty()));

        List<TaskContext> taskContextList = new ArrayList<>();
        taskContextList.add(taskContext);

        ColumnRefSet requiredOutputColumns = new ColumnRefSet(new ArrayList<>());

        new Expectations() {
            {
                context.getTaskContext();
                minTimes = 0;
                result = taskContextList;

                taskContext.getRequiredColumns();
                minTimes = 0;
                result = requiredOutputColumns;

                table.getPartitionColumnNames();
                result = scanColumnMap.values().stream().map(x -> x.getName()).collect(Collectors.toList());
            }
        };
        List<OptExpression> list = icebergRule.transform(scan, context);
        LogicalIcebergScanOperator op = ((LogicalIcebergScanOperator) list.get(0).getOp());
        Assert.assertEquals(op.getScanOptimzeOption().getCanUseAnyColumn(), false);
    }