This is an automated email from the ASF dual-hosted git repository.
924060929 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 794d514479e [feature](hive) Support static partition overwrite for
hive tables (#65991)
794d514479e is described below
commit 794d514479e038c883c1b2aa42069d17f98968a9
Author: Lijia Liu <[email protected]>
AuthorDate: Wed Jul 29 18:23:01 2026 +0800
[feature](hive) Support static partition overwrite for hive tables (#65991)
Add Hive static partition regression and FE plan tests, and fix
explicit-column INSERT cases to use the supported
PARTITION-before-column-list syntax.
---
.../nereids/analyzer/UnboundHiveTableSink.java | 49 ++-
.../nereids/analyzer/UnboundTableSinkCreator.java | 9 +-
.../doris/nereids/rules/analysis/BindSink.java | 92 ++++-
.../insert/InsertOverwriteTableCommand.java | 3 +-
.../trees/plans/commands/insert/InsertUtils.java | 17 +-
.../datasource/hive/HiveDDLAndDMLPlanTest.java | 104 ++++++
.../write/test_hive_write_static_partition.groovy | 383 +++++++++++++++++++++
7 files changed, 639 insertions(+), 18 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/UnboundHiveTableSink.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/UnboundHiveTableSink.java
index 4ffbc0230a0..134f2fc2e54 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/UnboundHiveTableSink.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/UnboundHiveTableSink.java
@@ -19,6 +19,7 @@ package org.apache.doris.nereids.analyzer;
import org.apache.doris.nereids.memo.GroupExpression;
import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.PlanType;
import org.apache.doris.nereids.trees.plans.commands.info.DMLCommandType;
@@ -26,8 +27,10 @@ import
org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
/**
@@ -35,14 +38,30 @@ import java.util.Optional;
*/
public class UnboundHiveTableSink<CHILD_TYPE extends Plan> extends
UnboundBaseExternalTableSink<CHILD_TYPE> {
+ // Static partition key-value pairs for INSERT [OVERWRITE] ... PARTITION
+ // (col='val', ...)
+ private final Map<String, Expression> staticPartitionKeyValues;
+
public UnboundHiveTableSink(List<String> nameParts, List<String> colNames,
List<String> hints,
List<String> partitions, CHILD_TYPE child) {
this(nameParts, colNames, hints, partitions, DMLCommandType.NONE,
- Optional.empty(), Optional.empty(), child);
+ Optional.empty(), Optional.empty(), child, null);
+ }
+
+ public UnboundHiveTableSink(List<String> nameParts,
+ List<String> colNames,
+ List<String> hints,
+ List<String> partitions,
+ DMLCommandType dmlCommandType,
+ Optional<GroupExpression> groupExpression,
+ Optional<LogicalProperties> logicalProperties,
+ CHILD_TYPE child) {
+ this(nameParts, colNames, hints, partitions, dmlCommandType,
+ groupExpression, logicalProperties, child, null);
}
/**
- * constructor
+ * constructor with static partition
*/
public UnboundHiveTableSink(List<String> nameParts,
List<String> colNames,
@@ -51,14 +70,21 @@ public class UnboundHiveTableSink<CHILD_TYPE extends Plan>
extends UnboundBaseEx
DMLCommandType dmlCommandType,
Optional<GroupExpression> groupExpression,
Optional<LogicalProperties> logicalProperties,
- CHILD_TYPE child) {
+ CHILD_TYPE child,
+ Map<String, Expression>
staticPartitionKeyValues) {
super(nameParts, PlanType.LOGICAL_UNBOUND_HIVE_TABLE_SINK,
ImmutableList.of(), groupExpression,
logicalProperties, colNames, dmlCommandType, child, hints,
partitions);
+ this.staticPartitionKeyValues = staticPartitionKeyValues != null
+ ? ImmutableMap.copyOf(staticPartitionKeyValues)
+ : null;
}
- @Override
- public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
- return visitor.visitUnboundHiveTableSink(this, context);
+ public Map<String, Expression> getStaticPartitionKeyValues() {
+ return staticPartitionKeyValues;
+ }
+
+ public boolean hasStaticPartition() {
+ return staticPartitionKeyValues != null &&
!staticPartitionKeyValues.isEmpty();
}
@Override
@@ -66,19 +92,24 @@ public class UnboundHiveTableSink<CHILD_TYPE extends Plan>
extends UnboundBaseEx
Preconditions.checkArgument(children.size() == 1,
"UnboundHiveTableSink only accepts one child");
return new UnboundHiveTableSink<>(nameParts, colNames, hints,
partitions,
- dmlCommandType, groupExpression, Optional.empty(),
children.get(0));
+ dmlCommandType, groupExpression, Optional.empty(),
children.get(0), staticPartitionKeyValues);
+ }
+
+ @Override
+ public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
+ return visitor.visitUnboundHiveTableSink(this, context);
}
@Override
public Plan withGroupExpression(Optional<GroupExpression> groupExpression)
{
return new UnboundHiveTableSink<>(nameParts, colNames, hints,
partitions,
- dmlCommandType, groupExpression,
Optional.of(getLogicalProperties()), child());
+ dmlCommandType, groupExpression,
Optional.of(getLogicalProperties()), child(), staticPartitionKeyValues);
}
@Override
public Plan withGroupExprLogicalPropChildren(Optional<GroupExpression>
groupExpression,
Optional<LogicalProperties>
logicalProperties, List<Plan> children) {
return new UnboundHiveTableSink<>(nameParts, colNames, hints,
partitions,
- dmlCommandType, groupExpression, logicalProperties,
children.get(0));
+ dmlCommandType, groupExpression, logicalProperties,
children.get(0), staticPartitionKeyValues);
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/UnboundTableSinkCreator.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/UnboundTableSinkCreator.java
index ff0cfc71264..ecadbf367d4 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/UnboundTableSinkCreator.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/UnboundTableSinkCreator.java
@@ -83,7 +83,7 @@ public class UnboundTableSinkCreator {
}
/**
- * create unbound sink for DML plan with static partition support for
Iceberg.
+ * create unbound sink for DML plan with static partition support for
Iceberg / Hive / MaxCompute.
*/
public static LogicalSink<? extends Plan>
createUnboundTableSink(List<String> nameParts,
List<String> colNames, List<String> hints, boolean
temporaryPartition, List<String> partitions,
@@ -98,7 +98,7 @@ public class UnboundTableSinkCreator {
Optional.empty(), plan);
} else if (curCatalog instanceof HMSExternalCatalog) {
return new UnboundHiveTableSink<>(nameParts, colNames, hints,
partitions,
- dmlCommandType, Optional.empty(), Optional.empty(), plan);
+ dmlCommandType, Optional.empty(), Optional.empty(), plan,
staticPartitionKeyValues);
} else if (curCatalog instanceof IcebergExternalCatalog) {
return new UnboundIcebergTableSink<>(nameParts, colNames, hints,
partitions,
dmlCommandType, Optional.empty(), Optional.empty(), plan,
staticPartitionKeyValues, false);
@@ -114,8 +114,7 @@ public class UnboundTableSinkCreator {
/**
* create unbound sink for DML plan with auto detect overwrite partition
enable
- * and static partition support for Iceberg.
- * TODO: staticPartitionKeyValues is only used for Iceberg, support other
catalog types in future.
+ * and static partition support for Iceberg / Hive / MaxCompute.
*/
public static LogicalSink<? extends Plan>
createUnboundTableSinkMaybeOverwrite(List<String> nameParts,
List<String> colNames, List<String> hints, boolean
temporaryPartition, List<String> partitions,
@@ -139,7 +138,7 @@ public class UnboundTableSinkCreator {
Optional.empty(), plan);
} else if (curCatalog instanceof HMSExternalCatalog &&
!isAutoDetectPartition) {
return new UnboundHiveTableSink<>(nameParts, colNames, hints,
partitions,
- dmlCommandType, Optional.empty(), Optional.empty(), plan);
+ dmlCommandType, Optional.empty(), Optional.empty(), plan,
staticPartitionKeyValues);
} else if (curCatalog instanceof IcebergExternalCatalog &&
!isAutoDetectPartition) {
return new UnboundIcebergTableSink<>(nameParts, colNames, hints,
partitions,
dmlCommandType, Optional.empty(), Optional.empty(), plan,
staticPartitionKeyValues, false);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java
index 8c580171b21..9a34aa5403f 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java
@@ -123,6 +123,7 @@ import org.apache.logging.log4j.Logger;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
@@ -690,11 +691,44 @@ public class BindSink implements AnalysisRuleFactory {
+ "(input format: " + inputFormat + "). LZO tables are
read-only in Doris.");
}
+ // Get static partition columns if present:
+ // INSERT [OVERWRITE] TABLE t PARTITION(col='val', ...) SELECT ...
+ Map<String, Expression> staticPartitions =
sink.getStaticPartitionKeyValues();
+ Set<String> staticPartitionColNames = staticPartitions != null
+ ? staticPartitions.keySet()
+ : Sets.newHashSet();
+ // Hive column/partition names are case-insensitive (stored lowercase
in HMS), so
+ // e.g. PARTITION(dt='21', DT='22') refers to the same column twice.
Detect such
+ // case-insensitive duplicates here and fail fast, instead of silently
letting a
+ // later value overwrite an earlier one in the case-insensitive
columnToOutput map.
+ // Use Locale.ROOT to keep the folding locale-independent (matching
how the Hive
+ // schema is loaded), otherwise e.g. the Turkish locale would fold 'I'
to 'ı' and
+ // fail to match a metadata column named 'id'.
+ Set<String> lowerStaticPartitionColNames = Sets.newHashSet();
+ for (String name : staticPartitionColNames) {
+ if
(!lowerStaticPartitionColNames.add(name.toLowerCase(Locale.ROOT))) {
+ throw new AnalysisException("Duplicate partition column: " +
name);
+ }
+ }
+
+ // Validate static partition columns against the hive table's
partition columns
+ if (sink.hasStaticPartition()) {
+ validateHiveStaticPartition(staticPartitions, table);
+ }
+
+ // Build bindColumns: exclude static partition columns from the
columns that
+ // need to come from SELECT, because their values come from the
PARTITION clause.
List<Column> bindColumns;
if (sink.getColNames().isEmpty()) {
- bindColumns =
table.getBaseSchema(true).stream().collect(ImmutableList.toImmutableList());
+ bindColumns = table.getBaseSchema(true).stream()
+ .filter(col ->
!lowerStaticPartitionColNames.contains(col.getName().toLowerCase(Locale.ROOT)))
+ .collect(ImmutableList.toImmutableList());
} else {
bindColumns = sink.getColNames().stream().map(cn -> {
+ if
(lowerStaticPartitionColNames.contains(cn.toLowerCase(Locale.ROOT))) {
+ throw new AnalysisException(String.format(
+ "column %s is a static partition column, should
not be in the insert column list", cn));
+ }
Column column = table.getColumn(cn);
if (column == null) {
throw new AnalysisException(String.format("column %s is
not found in table %s",
@@ -715,15 +749,71 @@ public class BindSink implements AnalysisRuleFactory {
Optional.empty(),
child);
// we need to insert all the columns of the target table
+ // (except the static partition columns which are supplied by the
PARTITION clause)
if (boundSink.getCols().size() != child.getOutput().size()) {
throw new AnalysisException("insert into cols should be
corresponding to the query output");
}
Map<String, NamedExpression> columnToOutput = getColumnToOutput(ctx,
table, false, false,
boundSink, child);
+
+ // For static partition columns, add constant expressions from the
PARTITION clause.
+ // This ensures every written row carries the static partition value,
so all rows
+ // land in the specified partition (and INSERT OVERWRITE only
overwrites that partition).
+ if (!staticPartitionColNames.isEmpty()) {
+ for (Map.Entry<String, Expression> entry :
staticPartitions.entrySet()) {
+ Expression valueExpr = entry.getValue();
+ Column column = table.getColumn(entry.getKey());
+ if (column != null) {
+ Expression castExpr = TypeCoercionUtils.castIfNotSameType(
+ valueExpr,
DataType.fromCatalogType(column.getType()));
+ // Use the canonical (lowercase) column name from the
table schema as both the
+ // map key and the alias name, so it aligns with
getOutputProjectByCoercion which
+ // looks up columnToOutput by table.getFullSchema() column
names.
+ columnToOutput.put(column.getName(), new Alias(castExpr,
column.getName()));
+ }
+ }
+ }
+
LogicalProject<?> fullOutputProject =
getOutputProjectByCoercion(table.getFullSchema(), child, columnToOutput);
return boundSink.withChildAndUpdateOutput(fullOutputProject);
}
+ /**
+ * Validate static partition specification for Hive table.
+ * The specified columns must be partition columns of the target hive
table,
+ * and each partition value must be a literal.
+ */
+ private void validateHiveStaticPartition(Map<String, Expression>
staticPartitions, HMSExternalTable table) {
+ if (staticPartitions == null || staticPartitions.isEmpty()) {
+ return;
+ }
+ Set<String> partitionColNames = table.getPartitionColumnNames();
+ if (partitionColNames.isEmpty()) {
+ throw new AnalysisException(
+ String.format("Table %s is not a partitioned table, cannot
use static partition syntax",
+ table.getName()));
+ }
+ // build a case-insensitive view of partition column names
+ Set<String> lowerPartitionColNames = Sets.newHashSet();
+ for (String name : partitionColNames) {
+ lowerPartitionColNames.add(name.toLowerCase());
+ }
+ for (Map.Entry<String, Expression> entry :
staticPartitions.entrySet()) {
+ String partitionColName = entry.getKey();
+ Expression partitionValue = entry.getValue();
+ if
(!lowerPartitionColNames.contains(partitionColName.toLowerCase())) {
+ throw new AnalysisException(
+ String.format("Unknown partition column '%s' in table
'%s'. Available partition columns: %s",
+ partitionColName, table.getName(),
partitionColNames));
+ }
+ if (!(partitionValue instanceof Literal)) {
+ throw new AnalysisException(
+ String.format("Partition value for column '%s' must be
a literal, but got: %s",
+ partitionColName, partitionValue));
+ }
+ }
+ }
+
private Plan
bindIcebergTableSink(MatchingContext<UnboundIcebergTableSink<Plan>> ctx) {
UnboundIcebergTableSink<?> sink = ctx.root;
Pair<IcebergExternalDatabase, IcebergExternalTable> pair =
bind(ctx.cascadesContext, sink);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java
index 7d5d0e49e77..9817db7796c 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java
@@ -375,7 +375,8 @@ public class InsertOverwriteTableCommand extends Command
implements NeedAuditEnc
false,
TPartialUpdateNewRowPolicy.APPEND,
sink.getDMLCommandType(),
- (LogicalPlan) (sink.child(0)));
+ (LogicalPlan) (sink.child(0)),
+ sink.getStaticPartitionKeyValues());
insertCtx = new HiveInsertCommandContext();
((HiveInsertCommandContext) insertCtx).setOverwrite(true);
} else if (logicalQuery instanceof UnboundIcebergTableSink) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertUtils.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertUtils.java
index fa5e34046d1..b555c8790db 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertUtils.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertUtils.java
@@ -99,6 +99,7 @@ import org.apache.commons.lang3.StringUtils;
import java.util.ArrayList;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
@@ -379,12 +380,24 @@ public class InsertUtils {
staticPartitions = ((UnboundIcebergTableSink<?>)
unboundLogicalSink).getStaticPartitionKeyValues();
} else if (unboundLogicalSink instanceof UnboundMaxComputeTableSink) {
staticPartitions = ((UnboundMaxComputeTableSink<?>)
unboundLogicalSink).getStaticPartitionKeyValues();
+ } else if (unboundLogicalSink instanceof UnboundHiveTableSink) {
+ staticPartitions = ((UnboundHiveTableSink<?>)
unboundLogicalSink).getStaticPartitionKeyValues();
}
if (staticPartitions != null && !staticPartitions.isEmpty()
&& CollectionUtils.isEmpty(unboundLogicalSink.getColNames())) {
- Set<String> staticPartitionColNames = staticPartitions.keySet();
+ // Static partition columns get their values from the PARTITION
clause instead of the
+ // inline VALUES list, so they must be excluded from the implicit
target schema before
+ // the column-count check below. Otherwise e.g. Hive table (v, dt)
with
+ // PARTITION(dt='x') VALUES (1) would be checked against 2 columns
and fail with
+ // "Column count doesn't match value count".
+ // Match case-insensitively via Locale.ROOT because partition
column names are
+ // case-insensitive (Hive stores them lowercase in HMS), avoiding
locale-dependent
+ // folding (e.g. the Turkish 'I').
+ Set<String> staticPartitionColNames =
staticPartitions.keySet().stream()
+ .map(name -> name.toLowerCase(Locale.ROOT))
+ .collect(Collectors.toSet());
columns = columns.stream()
- .filter(column ->
!staticPartitionColNames.contains(column.getName()))
+ .filter(column ->
!staticPartitionColNames.contains(column.getName().toLowerCase(Locale.ROOT)))
.collect(ImmutableList.toImmutableList());
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/hive/HiveDDLAndDMLPlanTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/hive/HiveDDLAndDMLPlanTest.java
index 5ff126ac30e..a7235b54420 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/hive/HiveDDLAndDMLPlanTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/hive/HiveDDLAndDMLPlanTest.java
@@ -573,6 +573,109 @@ public class HiveDDLAndDMLPlanTest extends
TestWithFeService {
checkPartTableSinkPlan(schema, partTargetTable,
physicalOverwriteSink2);
}
+ @Test
+ public void testInsertStaticPartitionPlanSql() throws Exception {
+ switchHive();
+ useDatabase(mockedDbName);
+ String insertTable = "static_partition_insert_table";
+ createTargetTable(insertTable);
+
+ List<Column> schema = new ArrayList<Column>() {
+ {
+ add(new Column("col1", PrimitiveType.INT));
+ add(new Column("pt1", PrimitiveType.VARCHAR, true));
+ add(new Column("pt2", PrimitiveType.STRING, true));
+ add(new Column("pt3", PrimitiveType.DATE, true));
+ }
+ };
+ Set<String> parts = new HashSet<String>() {
+ {
+ add("pt1");
+ add("pt2");
+ add("pt3");
+ }
+ };
+ mockTargetTable(schema, parts);
+ String partTargetTable = "part_" + insertTable;
+
+ String insertSql = "INSERT INTO " + partTargetTable
+ + " PARTITION(PT1='v1', pt2='v2', PT3='2020-03-13') SELECT 1";
+ PhysicalPlan physicalSink = getPhysicalPlan(insertSql,
+ new PhysicalProperties(new
DistributionSpecHiveTableSinkHashPartitioned()), false);
+ checkPartTableSinkPlan(schema, partTargetTable, physicalSink);
+
+ String insertOverwriteSql = "INSERT OVERWRITE TABLE " + partTargetTable
+ + " PARTITION(PT1='v1', PT2='v2', PT3='2020-03-13') SELECT 1";
+ PhysicalPlan physicalOverwriteSink =
getPhysicalPlan(insertOverwriteSql,
+ new PhysicalProperties(new
DistributionSpecHiveTableSinkHashPartitioned()), true);
+ checkPartTableSinkPlan(schema, partTargetTable, physicalOverwriteSink);
+
+ // Inline VALUES form without an explicit column list: the static
partition columns must be
+ // excluded from the implicit target schema, otherwise the single
VALUES row (col1 only)
+ // would be checked against all 4 columns and fail with "Column count
doesn't match".
+ String insertValuesSql = "INSERT INTO " + partTargetTable
+ + " PARTITION(PT1='v1', pt2='v2', PT3='2020-03-13') VALUES
(1)";
+ PhysicalPlan physicalValuesSink = getPhysicalPlan(insertValuesSql,
+ new PhysicalProperties(new
DistributionSpecHiveTableSinkHashPartitioned()), false);
+ checkPartTableSinkPlan(schema, partTargetTable, physicalValuesSink);
+
+ String insertOverwriteValuesSql = "INSERT OVERWRITE TABLE " +
partTargetTable
+ + " PARTITION(PT1='v1', PT2='v2', PT3='2020-03-13') VALUES
(1)";
+ PhysicalPlan physicalOverwriteValuesSink =
getPhysicalPlan(insertOverwriteValuesSql,
+ new PhysicalProperties(new
DistributionSpecHiveTableSinkHashPartitioned()), true);
+ checkPartTableSinkPlan(schema, partTargetTable,
physicalOverwriteValuesSink);
+ }
+
+ @Test
+ public void testInsertStaticPartitionErrorSql() throws Exception {
+ switchHive();
+ useDatabase(mockedDbName);
+ String insertTable = "static_partition_error_table";
+ createTargetTable(insertTable);
+
+ List<Column> schema = new ArrayList<Column>() {
+ {
+ add(new Column("col1", PrimitiveType.INT));
+ add(new Column("pt1", PrimitiveType.VARCHAR, true));
+ add(new Column("pt2", PrimitiveType.STRING, true));
+ add(new Column("pt3", PrimitiveType.DATE, true));
+ }
+ };
+ Set<String> parts = new HashSet<String>() {
+ {
+ add("pt1");
+ add("pt2");
+ add("pt3");
+ }
+ };
+ mockTargetTable(schema, parts);
+ String partTargetTable = "part_" + insertTable;
+
+ String nonLiteralPartitionSql = "INSERT INTO " + partTargetTable
+ + " PARTITION(pt1=concat('v', '1'), pt2='v2',
pt3='2020-03-13') SELECT 1";
+
ExceptionChecker.expectThrowsWithMsg(org.apache.doris.nereids.exceptions.AnalysisException.class,
+ "must be a literal",
+ () -> getPhysicalPlan(nonLiteralPartitionSql,
+ new PhysicalProperties(new
DistributionSpecHiveTableSinkHashPartitioned()), false));
+
+ String duplicateStaticPartitionSql = "INSERT INTO " + partTargetTable
+ + " PARTITION(PT1='v1', pt2='v2', pt3='2020-03-13') (col1,
pt1) SELECT 1, 'dup'";
+
ExceptionChecker.expectThrowsWithMsg(org.apache.doris.nereids.exceptions.AnalysisException.class,
+ "is a static partition column",
+ () -> getPhysicalPlan(duplicateStaticPartitionSql,
+ new PhysicalProperties(new
DistributionSpecHiveTableSinkHashPartitioned()), false));
+
+ // The same partition column specified twice with different casing is
a duplicate.
+ // Hive column names are case-insensitive, so pt1 and PT1 refer to the
same column and
+ // must be rejected instead of silently letting one value overwrite
the other.
+ String caseInsensitiveDupPartitionSql = "INSERT INTO " +
partTargetTable
+ + " PARTITION(pt1='v1', PT1='v2', pt2='v3', pt3='2020-03-13')
SELECT 1";
+
ExceptionChecker.expectThrowsWithMsg(org.apache.doris.nereids.exceptions.AnalysisException.class,
+ "Duplicate partition column",
+ () -> getPhysicalPlan(caseInsensitiveDupPartitionSql,
+ new PhysicalProperties(new
DistributionSpecHiveTableSinkHashPartitioned()), false));
+ }
+
private static void checkUnpartTableSinkPlan(List<Column> schema, String
unPartTargetTable, PhysicalPlan physicalSink) {
Assertions.assertSame(physicalSink.getType(),
PlanType.PHYSICAL_DISTRIBUTE);
// check exchange
@@ -742,3 +845,4 @@ public class HiveDDLAndDMLPlanTest extends
TestWithFeService {
checkedHiveCols.addAll(checkArrayCols);
}
}
+
diff --git
a/regression-test/suites/external_table_p0/hive/write/test_hive_write_static_partition.groovy
b/regression-test/suites/external_table_p0/hive/write/test_hive_write_static_partition.groovy
new file mode 100644
index 00000000000..9160a23c307
--- /dev/null
+++
b/regression-test/suites/external_table_p0/hive/write/test_hive_write_static_partition.groovy
@@ -0,0 +1,383 @@
+// 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.
+
+suite("test_hive_write_static_partition",
"p0,external,hive,external_docker,external_docker_hive") {
+
+ // Static partition insert:
+ // INSERT [OVERWRITE] TABLE t PARTITION(col='val', ...) SELECT ...
+ // The partition column value comes from the PARTITION clause instead of
the query output.
+ def testStaticPartitionWrite = { String catalog_name ->
+ String tableName = "hive_static_par_tbl"
+
+ hive_docker """ DROP TABLE IF EXISTS ${tableName}; """
+ hive_docker """
+ CREATE TABLE ${tableName} (
+ tag_value string,
+ user_id string,
+ ts int
+ )
+ PARTITIONED BY (ts_date string)
+ STORED AS parquet;
+ """
+ sql """ refresh catalog ${catalog_name}; """
+
+ // 1. INSERT INTO ... PARTITION(ts_date='2026-07-24') without listing
the partition column in SELECT.
+ // This used to fail with "insert into cols should be corresponding
to the query output".
+ sql """
+ INSERT INTO ${tableName} PARTITION(ts_date='2026-07-24')
+ SELECT 'tagA', 'u1', 100;
+ """
+ // 2. Append another row into the SAME partition.
+ sql """
+ INSERT INTO ${tableName} PARTITION(ts_date='2026-07-24')
+ SELECT 'tagB', 'u2', 200;
+ """
+ // 3. Insert into a DIFFERENT partition.
+ sql """
+ INSERT INTO ${tableName} PARTITION(ts_date='2026-07-25')
+ SELECT 'tagC', 'u3', 300;
+ """
+ sql """ refresh catalog ${catalog_name}; """
+
+ // Verify: 2 rows in partition 2026-07-24 and 1 row in 2026-07-25,
partition column filled correctly.
+ def all = sql """ select tag_value, user_id, ts, ts_date from
${tableName} order by ts_date, ts; """
+ assertEquals(3, all.size())
+ assertEquals("tagA", all[0][0]); assertEquals("2026-07-24", all[0][3])
+ assertEquals("tagB", all[1][0]); assertEquals("2026-07-24", all[1][3])
+ assertEquals("tagC", all[2][0]); assertEquals("2026-07-25", all[2][3])
+
+ // Partition pruning should work on the statically written partition
value.
+ def p24 = sql """ select tag_value from ${tableName} where
ts_date='2026-07-24' order by ts; """
+ assertEquals(2, p24.size())
+
+ // 4. INSERT OVERWRITE a single partition: it should only replace
2026-07-24, leaving 2026-07-25 intact.
+ sql """
+ INSERT OVERWRITE TABLE ${tableName} PARTITION(ts_date='2026-07-24')
+ SELECT 'tagX', 'u9', 999;
+ """
+ sql """ refresh catalog ${catalog_name}; """
+
+ def afterOverwrite = sql """ select tag_value, user_id, ts, ts_date
from ${tableName} order by ts_date, ts; """
+ assertEquals(2, afterOverwrite.size())
+ // partition 2026-07-24 replaced by a single row
+ assertEquals("tagX", afterOverwrite[0][0]); assertEquals(999,
afterOverwrite[0][2])
+ assertEquals("2026-07-24", afterOverwrite[0][3])
+ // partition 2026-07-25 untouched
+ assertEquals("tagC", afterOverwrite[1][0]); assertEquals("2026-07-25",
afterOverwrite[1][3])
+
+ // 5. Static partition combined with an explicit (non-partition)
column list.
+ sql """
+ INSERT INTO ${tableName} PARTITION(ts_date='2026-07-26')
(tag_value, user_id, ts)
+ SELECT 'tagD', 'u4', 400;
+ """
+ sql """ refresh catalog ${catalog_name}; """
+ def p26 = sql """ select tag_value, ts_date from ${tableName} where
ts_date='2026-07-26'; """
+ assertEquals(1, p26.size())
+ assertEquals("tagD", p26[0][0])
+
+ // 6. Partition column name matching should be case-insensitive.
+ sql """
+ INSERT INTO ${tableName} PARTITION(TS_DATE='2026-07-27')
+ SELECT 'tagE', 'u5', 500;
+ """
+ sql """ refresh catalog ${catalog_name}; """
+ def p27 = sql """ select tag_value, ts_date from ${tableName} where
ts_date='2026-07-27'; """
+ assertEquals(1, p27.size())
+ assertEquals("tagE", p27[0][0])
+ assertEquals("2026-07-27", p27[0][1])
+
+ // 7. Inline VALUES form (no explicit column list): partition column
value comes from the
+ // PARTITION clause, so VALUES only supplies the non-partition
columns.
+ sql """
+ INSERT INTO ${tableName} PARTITION(ts_date='2026-07-28')
+ VALUES ('tagF', 'u6', 600);
+ """
+ sql """ refresh catalog ${catalog_name}; """
+ def p28 = sql """ select tag_value, user_id, ts, ts_date from
${tableName} where ts_date='2026-07-28'; """
+ assertEquals(1, p28.size())
+ assertEquals("tagF", p28[0][0]); assertEquals("u6", p28[0][1])
+ assertEquals(600, p28[0][2]); assertEquals("2026-07-28", p28[0][3])
+
+ // 8. INSERT OVERWRITE with inline VALUES should only replace the
targeted partition.
+ sql """
+ INSERT OVERWRITE TABLE ${tableName} PARTITION(ts_date='2026-07-28')
+ VALUES ('tagG', 'u7', 700);
+ """
+ sql """ refresh catalog ${catalog_name}; """
+ def p28ow = sql """ select tag_value, ts, ts_date from ${tableName}
where ts_date='2026-07-28'; """
+ assertEquals(1, p28ow.size())
+ assertEquals("tagG", p28ow[0][0]); assertEquals(700, p28ow[0][1])
+ assertEquals("2026-07-28", p28ow[0][2])
+
+ hive_docker """ DROP TABLE IF EXISTS ${tableName}; """
+ }
+
+ def testStaticPartitionMultiColumnWrite = { String catalog_name ->
+ String tableName = "hive_static_multi_par_tbl"
+
+ hive_docker """ DROP TABLE IF EXISTS ${tableName}; """
+ hive_docker """
+ CREATE TABLE ${tableName} (
+ tag_value string,
+ user_id string,
+ ts int
+ )
+ PARTITIONED BY (dt string, region string)
+ STORED AS parquet;
+ """
+ sql """ refresh catalog ${catalog_name}; """
+
+ sql """
+ INSERT INTO ${tableName} PARTITION(dt='2026-07-24', region='bj')
+ SELECT 'tagM1', 'u10', 10;
+ """
+ sql """
+ INSERT INTO ${tableName} PARTITION(dt='2026-07-24', region='sh')
+ SELECT 'tagM2', 'u11', 20;
+ """
+ sql """
+ INSERT INTO ${tableName} PARTITION(dt='2026-07-25', region='bj')
+ SELECT 'tagM3', 'u12', 30;
+ """
+ sql """ refresh catalog ${catalog_name}; """
+
+ def beforeOverwrite = sql """
+ select tag_value, dt, region from ${tableName}
+ order by dt, region, tag_value;
+ """
+ assertEquals(3, beforeOverwrite.size())
+ assertEquals("tagM1", beforeOverwrite[0][0]);
assertEquals("2026-07-24", beforeOverwrite[0][1]); assertEquals("bj",
beforeOverwrite[0][2])
+ assertEquals("tagM2", beforeOverwrite[1][0]);
assertEquals("2026-07-24", beforeOverwrite[1][1]); assertEquals("sh",
beforeOverwrite[1][2])
+ assertEquals("tagM3", beforeOverwrite[2][0]);
assertEquals("2026-07-25", beforeOverwrite[2][1]); assertEquals("bj",
beforeOverwrite[2][2])
+
+ sql """
+ INSERT OVERWRITE TABLE ${tableName} PARTITION(dt='2026-07-24',
region='bj')
+ SELECT 'tagMX', 'u19', 999;
+ """
+ sql """ refresh catalog ${catalog_name}; """
+
+ def afterOverwrite = sql """
+ select tag_value, dt, region from ${tableName}
+ order by dt, region, tag_value;
+ """
+ assertEquals(3, afterOverwrite.size())
+ assertEquals("tagMX", afterOverwrite[0][0]);
assertEquals("2026-07-24", afterOverwrite[0][1]); assertEquals("bj",
afterOverwrite[0][2])
+ assertEquals("tagM2", afterOverwrite[1][0]);
assertEquals("2026-07-24", afterOverwrite[1][1]); assertEquals("sh",
afterOverwrite[1][2])
+ assertEquals("tagM3", afterOverwrite[2][0]);
assertEquals("2026-07-25", afterOverwrite[2][1]); assertEquals("bj",
afterOverwrite[2][2])
+
+ hive_docker """ DROP TABLE IF EXISTS ${tableName}; """
+ }
+
+ // Hybrid static/dynamic partition write: the PARTITION clause fixes only
a prefix of the
+ // partition columns (dt), while the remaining partition column (region)
stays dynamic and is
+ // sourced from the query output. A single statement can therefore emit
multiple dynamic
+ // partition values (region) under the same static prefix (dt). This
exercises a distinct sink
+ // routing path (synthesized dt slot + child-sourced region slot).
+ def testHybridStaticDynamicPartitionWrite = { String catalog_name ->
+ String tableName = "hive_hybrid_par_tbl"
+
+ hive_docker """ DROP TABLE IF EXISTS ${tableName}; """
+ hive_docker """
+ CREATE TABLE ${tableName} (
+ tag_value string,
+ user_id string,
+ ts int
+ )
+ PARTITIONED BY (dt string, region string)
+ STORED AS parquet;
+ """
+ sql """ refresh catalog ${catalog_name}; """
+
+ // 1. One INSERT with static dt='2026-07-24' emitting TWO dynamic
regions (bj, sh).
+ // The SELECT only supplies non-static columns: tag_value, user_id,
ts, region.
+ sql """
+ INSERT INTO ${tableName} PARTITION(dt='2026-07-24')
+ SELECT 'h_bj', 'u1', 10, 'bj'
+ UNION ALL
+ SELECT 'h_sh', 'u2', 20, 'sh';
+ """
+ // 2. A sibling static prefix dt='2026-07-25', also emitting two
dynamic regions.
+ sql """
+ INSERT INTO ${tableName} PARTITION(dt='2026-07-25')
+ SELECT 's_bj', 'u3', 30, 'bj'
+ UNION ALL
+ SELECT 's_gz', 'u4', 40, 'gz';
+ """
+ sql """ refresh catalog ${catalog_name}; """
+
+ def all = sql """
+ select tag_value, user_id, ts, dt, region from ${tableName}
+ order by dt, region;
+ """
+ assertEquals(4, all.size())
+ assertEquals("h_bj", all[0][0]); assertEquals(10, all[0][2]);
assertEquals("2026-07-24", all[0][3]); assertEquals("bj", all[0][4])
+ assertEquals("h_sh", all[1][0]); assertEquals(20, all[1][2]);
assertEquals("2026-07-24", all[1][3]); assertEquals("sh", all[1][4])
+ assertEquals("s_bj", all[2][0]); assertEquals(30, all[2][2]);
assertEquals("2026-07-25", all[2][3]); assertEquals("bj", all[2][4])
+ assertEquals("s_gz", all[3][0]); assertEquals(40, all[3][2]);
assertEquals("2026-07-25", all[3][3]); assertEquals("gz", all[3][4])
+
+ // Partition pruning should work on both the static and the dynamic
partition columns.
+ def bjRows = sql """ select tag_value from ${tableName} where
region='bj' order by dt; """
+ assertEquals(2, bjRows.size())
+ assertEquals("h_bj", bjRows[0][0]); assertEquals("s_bj", bjRows[1][0])
+
+ // 3. Hybrid INSERT OVERWRITE under static dt='2026-07-24' re-emitting
the same two regions.
+ // It must only replace the dt='2026-07-24' partitions and leave
the dt='2026-07-25'
+ // sibling prefix completely untouched.
+ sql """
+ INSERT OVERWRITE TABLE ${tableName} PARTITION(dt='2026-07-24')
+ SELECT 'o_bj', 'u5', 50, 'bj'
+ UNION ALL
+ SELECT 'o_sh', 'u6', 60, 'sh';
+ """
+ sql """ refresh catalog ${catalog_name}; """
+
+ def afterOverwrite = sql """
+ select tag_value, user_id, ts, dt, region from ${tableName}
+ order by dt, region;
+ """
+ assertEquals(4, afterOverwrite.size())
+ // dt='2026-07-24' partitions replaced with new values.
+ assertEquals("o_bj", afterOverwrite[0][0]); assertEquals(50,
afterOverwrite[0][2]); assertEquals("2026-07-24", afterOverwrite[0][3]);
assertEquals("bj", afterOverwrite[0][4])
+ assertEquals("o_sh", afterOverwrite[1][0]); assertEquals(60,
afterOverwrite[1][2]); assertEquals("2026-07-24", afterOverwrite[1][3]);
assertEquals("sh", afterOverwrite[1][4])
+ // dt='2026-07-25' sibling prefix untouched.
+ assertEquals("s_bj", afterOverwrite[2][0]); assertEquals(30,
afterOverwrite[2][2]); assertEquals("2026-07-25", afterOverwrite[2][3]);
assertEquals("bj", afterOverwrite[2][4])
+ assertEquals("s_gz", afterOverwrite[3][0]); assertEquals(40,
afterOverwrite[3][2]); assertEquals("2026-07-25", afterOverwrite[3][3]);
assertEquals("gz", afterOverwrite[3][4])
+
+ hive_docker """ DROP TABLE IF EXISTS ${tableName}; """
+ }
+
+ // Error cases for static partition validation.
+ def testStaticPartitionErrors = { String catalog_name ->
+ String tableName = "hive_static_par_err_tbl"
+ String nonParTableName = "hive_static_par_nonpar_tbl"
+
+ hive_docker """ DROP TABLE IF EXISTS ${tableName}; """
+ hive_docker """
+ CREATE TABLE ${tableName} (
+ tag_value string,
+ user_id string,
+ ts int
+ )
+ PARTITIONED BY (ts_date string)
+ STORED AS parquet;
+ """
+ hive_docker """ DROP TABLE IF EXISTS ${nonParTableName}; """
+ hive_docker """
+ CREATE TABLE ${nonParTableName} (
+ tag_value string,
+ user_id string
+ )
+ STORED AS parquet;
+ """
+ sql """ refresh catalog ${catalog_name}; """
+
+ // 5.1 partition column not exists
+ test {
+ sql """
+ INSERT INTO ${tableName} PARTITION(not_exist_col='2026-07-24')
+ SELECT 'tagA', 'u1', 100;
+ """
+ exception "Unknown partition column"
+ }
+
+ // 5.2 static partition column also appears in the insert column list
+ test {
+ sql """
+ INSERT INTO ${tableName} PARTITION(ts_date='2026-07-24')
(tag_value, user_id, ts, ts_date)
+ SELECT 'tagA', 'u1', 100, '2026-07-24';
+ """
+ exception "is a static partition column"
+ }
+
+ // 5.3 use static partition syntax on a non-partitioned table
+ test {
+ sql """
+ INSERT INTO ${nonParTableName} PARTITION(ts_date='2026-07-24')
+ SELECT 'tagA', 'u1';
+ """
+ exception "is not a partitioned table"
+ }
+
+ // 5.4 partition value must be a literal instead of an expression
+ test {
+ sql """
+ INSERT INTO ${tableName} PARTITION(ts_date=concat('2026-07',
'-24'))
+ SELECT 'tagA', 'u1', 100;
+ """
+ exception "must be a literal"
+ }
+
+ // 5.5 case-insensitive partition column name should also be
recognized in insert column list validation
+ test {
+ sql """
+ INSERT INTO ${tableName} PARTITION(TS_DATE='2026-07-24')
(tag_value, user_id, ts, ts_date)
+ SELECT 'tagA', 'u1', 100, '2026-07-24';
+ """
+ exception "is a static partition column"
+ }
+
+ // 5.6 the same partition column specified twice with different casing
is a duplicate.
+ // Hive column names are case-insensitive, so ts_date and TS_DATE
are the same column.
+ // This must fail fast instead of silently letting one value
overwrite the other.
+ test {
+ sql """
+ INSERT INTO ${tableName} PARTITION(ts_date='2026-07-24',
TS_DATE='2026-07-25')
+ SELECT 'tagA', 'u1', 100;
+ """
+ exception "Duplicate partition column"
+ }
+
+ hive_docker """ DROP TABLE IF EXISTS ${tableName}; """
+ hive_docker """ DROP TABLE IF EXISTS ${nonParTableName}; """
+ }
+
+ String enabled = context.config.otherConfigs.get("enableHiveTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable Hive test.")
+ return;
+ }
+
+ for (String hivePrefix : ["hive2", "hive3"]) {
+ setHivePrefix(hivePrefix)
+ try {
+ String hms_port = context.config.otherConfigs.get(hivePrefix +
"HmsPort")
+ String hdfs_port = context.config.otherConfigs.get(hivePrefix +
"HdfsPort")
+ String catalog_name = "test_${hivePrefix}_write_static_partition"
+ String externalEnvIp =
context.config.otherConfigs.get("externalEnvIp")
+
+ sql """drop catalog if exists ${catalog_name}"""
+ sql """create catalog if not exists ${catalog_name} properties (
+ 'type'='hms',
+ 'hive.metastore.uris' =
'thrift://${externalEnvIp}:${hms_port}',
+ 'fs.defaultFS' = 'hdfs://${externalEnvIp}:${hdfs_port}'
+ );"""
+ sql """use `${catalog_name}`.`write_test`"""
+ logger.info("hive sql: use `write_test`")
+ hive_docker """use `write_test`"""
+
+ sql """set enable_fallback_to_original_planner=false;"""
+
+ testStaticPartitionWrite(catalog_name)
+ testStaticPartitionMultiColumnWrite(catalog_name)
+ testHybridStaticDynamicPartitionWrite(catalog_name)
+ testStaticPartitionErrors(catalog_name)
+
+ sql """drop catalog if exists ${catalog_name}"""
+ } finally {
+ }
+ }
+}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
