apache · sparkyen · Jul 9, 2026
@@ -147,3 +147,8 @@ compile_commands.json
 .github
 
 .worktrees/
+
+# Keep horn4j chunks tracked while the reconstructed jar remains ignored by
+# the global *.jar rule. Apache's large-file warning checks individual files.
+!/fe/custom_libs/horn4j-1.0.0-SNAPSHOT.jar.parts/
+!/fe/custom_libs/horn4j-1.0.0-SNAPSHOT.jar.parts/*
@@ -571,6 +571,7 @@ if [[ "${BUILD_FE}" -eq 1 ]]; then
     modules+=("fe-common")
     modules+=("fe-extension-spi")
     modules+=("fe-extension-loader")
+    modules+=("horn4j")
     modules+=("fe-core")
     if [[ "${WITH_TDE_DIR}" != "" ]]; then
         modules+=("fe-${WITH_TDE_DIR}")

@@ -97,6 +97,11 @@ under the License.
             <artifactId>fe-extension-loader</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>

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

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

@@ -0,0 +1,159 @@
+// 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.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 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 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
+ * output-rebuild phases.
+ */
+public class HornOptimizationContext {
+
+    private final CascadesContext cascadesContext;
+
+    /** Table metadata + statistics. */
+    private TTableBatch tableBatch = new TTableBatch();
+
+    /** Horn optimizer configuration. */
+    private TQueryOptionConfig queryOptionConfig;
+
+    /**
+     * forward 每个 LogicalOlapScan 的裁剪后分区集，按 relationId 索引（Horn 只透传 relationId，
+     * 不进 thrift）。backward DorisPhysicalPlanBuilder.buildScan 用它重建 scan 分区。
+     */
+    private final Map<Long, List<Long>> scanPrunedPartitions = new HashMap<>();
+
+    /**
+     * backward 阶段 horn scalar_unique_id → 反译出的 Doris Slot 全局映射，保证上下游 plan 节点的
+     * slot 引用一致（{@link org.apache.doris.horn.horn2doris.DorisPhysicalPlanBuilder} 写入、
+     * {@link org.apache.doris.horn.horn2doris.DorisExpressionBuilder} 反译 ColumnRef 时按 uid 查）。
+     */
+    private final Map<Long, Slot> scalarUidToSlot = new HashMap<>();
+
+    /**
+     * backward 阶段 horn agg fn 的 scalar_unique_id → 反译出的 Doris AggregateFunction，专给两阶段
+     * agg 用：local agg put、global agg 按 uid get 回同一实例（Doris 用引用关联 phase 间的同一 fn）。
+     */
+    private final Map<Long, AggregateFunction> scalarUidToAggFunc = new HashMap<>();
+
+    public HornOptimizationContext(CascadesContext cascadesContext) {
+        this.cascadesContext = cascadesContext;
+        this.queryOptionConfig = buildDefaultQueryOptionConfig();
+        applyLeadingHint(this.queryOptionConfig);
+    }
+
+    /**
+     * 把 SQL 的 {@code LEADING(...)} hint 转发给 horn，让内部 {@code XformLeadingJoin} 按指定顺序重排 join。
+     */
+    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 估算）——否则缺 stats 的表
+        // 会大量 fallback Nereids。严格 stats 后续由 session var 控制。
+        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
+        config.setEnable_horn_profile(cascadesContext.getConnectContext()
+                .getSessionVariable().enableHornProfile);
+        config.setExplain_level(getExplainLevel());
+        return config;
+    }
+}
@@ -0,0 +1,109 @@
+// 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.doris.nereids.CascadesContext;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalPlan;
+
+import org.apache.horn4j.HornNative;
+import org.apache.horn4j.thrift.TFlattenedExpression;
+import org.apache.horn4j.thrift.THornOptimizeResult;
+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.
+ *
+ * Flow: Doris LogicalPlan → TFlattenedExpression → horn4j JNI → Horn (in-process)
+ *       → THornOptimizeResult → Doris PhysicalPlan.
+ */
+public class HornOptimizer {
+
+    private static final Logger LOG = LogManager.getLogger(HornOptimizer.class);
+
+    private final CascadesContext cascadesContext;
+    private String hornExplainString;
+    private String fallbackReason;
+
+    public HornOptimizer(CascadesContext cascadesContext) {
+        this.cascadesContext = cascadesContext;
+    }
+
+    public String getHornExplainString() {
+        return hornExplainString;
+    }
+
+    public String getFallbackReason() {
+        return fallbackReason;
+    }
+
+    public PhysicalPlan optimize(LogicalPlan logicalPlan) throws UserException {
+        HornOptimizationContext hornCtx = new HornOptimizationContext(cascadesContext);
+
+        try {
+            // Step 1: Doris LogicalPlan → TFlattenedExpression
+            LOG.info("Horn CBO: translating Doris plan to Horn input");
+
+            HornLogicalPlanThriftBuilder planVisitor = new HornLogicalPlanThriftBuilder(hornCtx);
+            TFlattenedExpression plan = planVisitor.translate(logicalPlan);
+
+            TSerializer serializer = new TSerializer();
+            byte[] planBytes = serializer.serialize(plan);
+            byte[] tableBatchBytes = serializer.serialize(hornCtx.getTableBatch());
+            byte[] queryOptionBytes = serializer.serialize(hornCtx.getQueryOptionConfig());
+
+            // Step 2: call Horn in-process via JNI (horn4j).
+            LOG.info("Horn CBO: calling Horn via horn4j JNI");
+
+            byte[] resultBytes = new HornNative().optimize(
+                    planBytes, tableBatchBytes, queryOptionBytes);
+
+            // Step 3: Horn output → Doris PhysicalPlan.
+            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);
+            }
+
+            DorisPhysicalPlanBuilder builder =
+                    new DorisPhysicalPlanBuilder(hornCtx);
+            PhysicalPlan physicalPlan = builder.build(hornResult.getTfexpr());
+
+            LOG.info("Horn CBO: optimization complete");
+            return physicalPlan;
+
+        } catch (Exception e) {
+            fallbackReason = e.getMessage();
+            LOG.warn("Horn CBO failed, falling back to Nereids: {}", fallbackReason, e);
+            return null;
+        }
+    }
+
+}
@@ -0,0 +1,40 @@
+// 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 侧的
+ * {@code Horn2DorisUtils}）。跟具体 visitor 实例状态无关的小工具放这里复用。
+ */
+public final class Doris2HornUtils {
+
+    private Doris2HornUtils() {
+    }
+
+    /**
+     * 剥掉一层 {@link Alias} 取内部表达式（Alias 在 horn wire 上透明，exprId/name 由输出列
+     * SlotReference 携带）；非 Alias 原样返回。
+     */
+    public static Expression unwrapExpr(Expression expr) {
+        return (expr instanceof Alias) ? ((Alias) expr).child() : expr;
+    }
+}