HornOptimizer.java

// 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.horn;

import org.apache.doris.common.UserException;
import org.apache.doris.horn.doris2horn.HornLogicalPlanThriftBuilder;
import org.apache.doris.horn.horn2doris.DorisPhysicalPlanBuilder;
import org.apache.horn4j.thrift.TFlattenedExpression;
import org.apache.horn4j.thrift.THornOptimizeProfile;
import org.apache.horn4j.thrift.THornOptimizeResult;
import org.apache.horn4j.HornNative;
import org.apache.doris.metric.HornMetric;
import org.apache.doris.metric.MetricRepo;
import org.apache.doris.nereids.CascadesContext;
import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
import org.apache.doris.nereids.trees.plans.physical.PhysicalPlan;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.thrift.TDeserializer;
import org.apache.thrift.TSerializer;

/** Doris-side entry to the Horn CBO optimizer */
public class HornOptimizer {

    private static final Logger LOG = LogManager.getLogger(HornOptimizer.class);

    /** Horn ������������������������������ Horn ��������������������������������� ��� Horn ������ ��� ��������������� */
    public enum HornFailureKind {
        // A: Doris���Horn ������������������ ��� ��� translate_plan_error���Doris ������������������������
        HORN_FORWARD_ERROR,
        // B: Horn ������������������HornNotHandleError/External��� ��� ��� fallback��������� kernel countHornFallBack���
        HORN_NOT_HANDLE,
        // C: Horn ������������������������HornOptimizerError��� ��� ��� error��������� kernel countHornError���
        HORN_OPTIMIZE_ERROR,
        // D: Horn���Doris ������������������ ��� ��� translate_plan_error���Doris ������������������������
        HORN_BACKWARD_ERROR,
        // ���������native ���������������������������������������JNI ��������� / ������ std ������ / ��������������� ��� ��� error
        UNKNOWN
    }

    private final CascadesContext cascadesContext;
    private String hornExplainString;
    private String fallbackReason;
    // ������ optimize() ��������������� null������������������ NereidsPlanner ���������������������������������
    private HornFailureKind failureKind;

    public HornOptimizer(CascadesContext cascadesContext) {
        this.cascadesContext = cascadesContext;
    }

    public String getHornExplainString() {
        return hornExplainString;
    }

    public String getFallbackReason() {
        return fallbackReason;
    }

    public HornFailureKind getFailureKind() {
        return failureKind;
    }

    public PhysicalPlan optimize(LogicalPlan logicalPlan) throws UserException {
        HornOptimizationContext hornCtx = new HornOptimizationContext(cascadesContext);

        // ������ A���Doris LogicalPlan ��� TFlattenedExpression ��� ������������
        byte[] planBytes;
        byte[] tableBatchBytes;
        byte[] queryOptionBytes;
        try {
            LOG.info("Horn CBO: translating Doris plan to Horn input");
            HornLogicalPlanThriftBuilder planVisitor = new HornLogicalPlanThriftBuilder(hornCtx);
            TFlattenedExpression plan = planVisitor.translate(logicalPlan);

            TSerializer serializer = new TSerializer();
            planBytes = serializer.serialize(plan);
            tableBatchBytes = serializer.serialize(hornCtx.getTableBatch());
            queryOptionBytes = serializer.serialize(hornCtx.getQueryOptionConfig());
        } catch (Exception e) {
            failureKind = HornFailureKind.HORN_FORWARD_ERROR;
            fallbackReason = e.getMessage();
            LOG.info("Horn CBO: forward-translation failed, fallback: {}", fallbackReason);
            return null;
        }

        // ������ B/C���JNI ��� Horn native���in-process horn4j������
        byte[] resultBytes;
        try {
            LOG.info("Horn CBO: calling Horn via horn4j JNI");
            resultBytes = new HornNative().optimize(planBytes, tableBatchBytes, queryOptionBytes);
        } catch (Exception e) {
            // ������ kernel HornError ���������������������������������horn_error.h GetErrorTypeString ��������������� message���
            String msg = e.getMessage();
            if (msg != null && msg.contains("HornNotHandle")) {
                failureKind = HornFailureKind.HORN_NOT_HANDLE;
            } else if (msg != null && msg.contains("HornOptimizerError")) {
                failureKind = HornFailureKind.HORN_OPTIMIZE_ERROR;
            } else {
                failureKind = HornFailureKind.UNKNOWN;
            }
            fallbackReason = msg;
            LOG.warn("Horn CBO: native failed ({}), fallback: {}", failureKind, fallbackReason);
            return null;
        }

        // ������ D��������������� + ������ profile + Horn output ��� Doris PhysicalPlan���
        try {
            LOG.info("Horn CBO: rebuilding Doris plan from Horn output");
            THornOptimizeResult hornResult = new THornOptimizeResult();
            TDeserializer deserializer = new TDeserializer();
            deserializer.deserialize(hornResult, resultBytes);

            hornExplainString = hornResult.getExplain_string();
            if (hornExplainString != null) {
                LOG.info("Horn CBO explain:\n{}", hornExplainString);
            }

            consumeProfile(hornResult);

            DorisPhysicalPlanBuilder builder = new DorisPhysicalPlanBuilder(hornCtx);
            PhysicalPlan physicalPlan = builder.build(hornResult.getTfexpr());

            LOG.info("Horn CBO: optimization complete");
            return physicalPlan;
        } catch (Exception e) {
            failureKind = HornFailureKind.HORN_BACKWARD_ERROR;
            // ��������������� horn C++ ������������������������getHornExplainString() ���������������������
            fallbackReason = e.getMessage();
            LOG.warn("Horn CBO: backward-translation failed, fallback: {}", fallbackReason, e);
            return null;
        }
    }

    /** ������ Horn profile ��� ������ FE ���������������profile ������������������������������ session ��������� */
    private static void consumeProfile(THornOptimizeResult hornResult) {
        if (hornResult.isSetProfile() && MetricRepo.isInit) {
            THornOptimizeProfile p = hornResult.getProfile();
            HornMetric.GAUGE_HORN_OPTIMIZE_LATENCY_NS.setValue(p.getHorn_optimize_latency());          // #6
            HornMetric.GAUGE_HORN_CASCADE_GROUP_COUNT.setValue(p.getCascade_group_count());            // #7
            HornMetric.GAUGE_HORN_CASCADE_GROUP_EXPR_COUNT.setValue(
                    p.getCascade_group_expression_count());                                            // #8
            HornMetric.GAUGE_HORN_CASCADE_SAFE_TO_PRUNE_RATIO.setValue(
                    p.getCascade_safe_to_prune_ratio());                                               // #9
            HornMetric.GAUGE_HORN_SCHEDULER_JOB_COUNT.setValue(p.getScheduler_job_count());            // #10
            HornMetric.GAUGE_HORN_DPHYPER_CSG_CMP_COUNT.setValue(p.getDphyper_csg_cmp_count());        // #11
            HornMetric.GAUGE_HORN_DPHYPER_ENUM_SUCCESS_RATIO.setValue(
                    p.getDphyper_enumeration_successful_ratio());                                      // #12
        }
    }

}