apache · sparkyen · Jul 16, 2026
@@ -588,6 +588,7 @@ FE_MODULES=''
 modules=("")
 if [[ "${BUILD_FE}" -eq 1 ]]; then
     modules+=("fe-common")
+    modules+=("horn4j")
     modules+=("fe-core")
     if [[ "${WITH_TDE_DIR}" != "" ]]; then
         modules+=("fe-${WITH_TDE_DIR}")

@@ -75,4 +75,8 @@ under the License.
     <!-- ignore gensrc/thrift/ExternalTableSchema.thrift -->
     <suppress files=".*thrift/schema/external/.*" checks=".*"/>
     <suppress files="software/amazon/awssdk/core/client/builder/SdkDefaultClientBuilder.java" checks=".*"/>
+
+    <!-- horn optimizer files -->
+    <suppress files="org[\\/]apache[\\/]doris[\\/]horn[\\/]" checks="[a-zA-Z0-9]*"/>
+    <suppress files="org[\\/]apache[\\/]doris[\\/]metric[\\/]HornMetric\.java" checks="[a-zA-Z0-9]*"/>
 </suppressions>
@@ -138,6 +138,11 @@ under the License.
             <artifactId>fe-authentication-role-mapping</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.horn4j</groupId>
+            <artifactId>horn4j</artifactId>
+            <version>1.0.0-SNAPSHOT</version>
+        </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-devtools</artifactId>

@@ -285,6 +285,7 @@ HELP: 'HELP';
 HISTOGRAM: 'HISTOGRAM';
 HLL: 'HLL';
 HLL_UNION: 'HLL_UNION';
+HORN: 'HORN';
 HOSTNAME: 'HOSTNAME';
 HOTSPOT: 'HOTSPOT';
 HOUR: 'HOUR';

@@ -1177,6 +1177,7 @@ planType
     | SHAPE
     | MEMO
     | DISTRIBUTED
+    | HORN
     | ALL // default type
     ;
 
@@ -2160,6 +2161,7 @@ nonReserved
     | HINT_START
     | HISTOGRAM
     | HLL_UNION
+    | HORN
     | HOSTNAME
     | HOTSPOT
     | HOUR

@@ -0,0 +1,140 @@
+// 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.horn4j.thrift.TEngineType;
+import org.apache.horn4j.thrift.TExplainLevel;
+import org.apache.horn4j.thrift.TQueryOptionConfig;
+import org.apache.horn4j.thrift.TTable;
+import org.apache.horn4j.thrift.TTableBatch;
+import org.apache.doris.nereids.CascadesContext;
+import org.apache.doris.nereids.hint.LeadingHint;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/** Horn optimization context shared across the input-translation, JNI, and */
+public class HornOptimizationContext {
+
+    private final CascadesContext cascadesContext;
+
+    /** Table metadata + statistics */
+    private TTableBatch tableBatch = new TTableBatch();
+
+    /** Horn optimizer configuration */
+    private TQueryOptionConfig queryOptionConfig;
+
+    /** forward 每个 LogicalOlapScan 的裁剪后分区集,按 relationId(= thrift */
+    private final Map<Long, List<Long>> scanPrunedPartitions = new HashMap<>();
+
+    /** backward 阶段 horn 端 scalar_unique_id → 反译产生的 Doris Slot 的全局映射， */
+    private final Map<Long, Slot> scalarUidToSlot = new HashMap<>();
+
+    /** backward 阶段 horn 端 agg fn 的 scalar_unique_id → 反译出的 Doris */
+    private final Map<Long, AggregateFunction> scalarUidToAggFunc = new HashMap<>();
+
+    public HornOptimizationContext(CascadesContext cascadesContext) {
+        this.cascadesContext = cascadesContext;
+        this.queryOptionConfig = buildDefaultQueryOptionConfig();
+        applyLeadingHint(this.queryOptionConfig);
+    }
+
+    /** 把 SQL 中 {@code /*+ LEADING(t1 t2 t3) */} hint 转发给 horn —— 让 horn 内部 */
+    private void applyLeadingHint(TQueryOptionConfig config) {
+        if (!cascadesContext.isLeadingJoin()) {
+            return;
+        }
+        LeadingHint leading = (LeadingHint) cascadesContext.getHintMap().get("Leading");
+        if (leading != null && !leading.getTablelist().isEmpty()) {
+
+            config.setJoin_leading_string(new ArrayList<>(leading.getTablelist()));
+        }
+    }
+
+    public CascadesContext getCascadesContext() {
+        return cascadesContext;
+    }
+
+    public TTableBatch getTableBatch() {
+        return tableBatch;
+    }
+
+    public void setTableBatch(TTableBatch tableBatch) {
+        this.tableBatch = tableBatch;
+    }
+
+    public void putScanPrunedPartitions(long relationId, List<Long> prunedPartitionIds) {
+        scanPrunedPartitions.put(relationId, prunedPartitionIds);
+    }
+
+    public List<Long> getScanPrunedPartitions(long relationId) {
+        return scanPrunedPartitions.get(relationId);
+    }
+
+    public void addTable(TTable ttable) {
+        if (tableBatch.getTables() == null) {
+            tableBatch.setTables(new ArrayList<>());
+        }
+        tableBatch.getTables().add(ttable);
+    }
+
+    public Map<Long, Slot> getScalarUidToSlot() {
+        return scalarUidToSlot;
+    }
+
+    public Map<Long, AggregateFunction> getScalarUidToAggFunc() {
+        return scalarUidToAggFunc;
+    }
+
+    public TQueryOptionConfig getQueryOptionConfig() {
+        return queryOptionConfig;
+    }
+
+    public TExplainLevel getExplainLevel() {
+        return TExplainLevel.EXTRACTED_PLAN;
+    }
+
+    private TQueryOptionConfig buildDefaultQueryOptionConfig() {
+        TQueryOptionConfig config = new TQueryOptionConfig();
+        config.setHorn_dphyper_enable(true);
+        config.setHorn_statistics_default_selectivity(0.4);
+        config.setHorn_statistics_damping_factor_filter(0.75);
+        config.setHorn_statistics_damping_factor_groupby(0.75);
+        config.setHorn_statistics_damping_factor_join(0.1);
+        // 默认允许无统计的表也走 horn（用 default selectivity 等估算）
+        config.setEnable_horn_no_statistics_support(true);
+        config.setLocal_time_zone(cascadesContext.getConnectContext()
+                .getSessionVariable().getTimeZone());
+        config.setMt_dop(1);
+        config.setEngine_type(TEngineType.kDoris);
+        boolean detailInfo = cascadesContext.getConnectContext()
+                .getSessionVariable().enableHornDetailInfo;
+        config.setEnable_horn_cascade_log(detailInfo);
+        config.setEnable_horn_cascade_detail_log(detailInfo);
+        config.setEnable_horn_stats_log(detailInfo);
+        config.setHorn_print_memo_after_optimization(detailInfo);  // detail 模式下 dump memo（调试 cascade 用）
+        config.setEnable_horn_profile(cascadesContext.getConnectContext()
+                .getSessionVariable().enableHornProfile);
+        config.setExplain_level(getExplainLevel());
+        return config;
+    }
+}
@@ -0,0 +1,167 @@
+// 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
+        }
+    }
+
+}
@@ -0,0 +1,33 @@
+// 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.doris2horn;
+
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.Expression;
+
+/** doris2horn 正向翻译路径下的纯静态辅助方法集中点（对称 horn2doris 侧的 */
+public final class Doris2HornUtils {
+
+    private Doris2HornUtils() {
+    }
+
+    /** 剥掉一层 {@link Alias} 取内部表达式：Alias 在 horn wire 上是透明的（exprId/name */
+    public static Expression unwrapExpr(Expression expr) {
+        return (expr instanceof Alias) ? ((Alias) expr).child() : expr;
+    }
+}