This is an automated email from the ASF dual-hosted git repository.
morningman 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 f8a4eba21b8 [improvement](iceberg) Route partitioned Iceberg writes by
the transformed partition value (#68921)
f8a4eba21b8 is described below
commit f8a4eba21b8b63e33ddf632ef9127889e994c881
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Thu Oct 15 14:50:32 2026 +0800
[improvement](iceberg) Route partitioned Iceberg writes by the transformed
partition value (#68921)
### What problem does this PR solve?
Issue Number: None
Related PR: #66685
Problem Summary:
**In short.** This ports the Iceberg part of branch-4.x #66685 to
master: an INSERT into a partitioned Iceberg table is now
hash-distributed by the transformed partition value, so each partition
is written by one writer. On master today the rows are spread randomly,
and a single INSERT that touches more than 128 partitions fails.
**Background.** Every writer of an Iceberg sink keeps one open file per
partition it has seen, capped by the BE config
`table_sink_partition_write_max_partition_nums_per_writer` (default
128). branch-4.x #66685 routes partitioned Iceberg writes through an
Iceberg-transform exchange. The FE computes the active partition spec's
transforms, the BE applies them before hashing, and all rows of one
partition reach one writer. Master's Iceberg connector only declares
`requiresParallelWrite`, so the engine falls back to random
distribution. The external sink exchange that master already has
(connector-defined partition functions, used by Paimon's fixed buckets)
has no Iceberg function.
**The problem, and what it cost.** With random distribution every writer
receives rows of every partition:
| | master today | with this PR |
|---|---|---|
| `INSERT ... SELECT number, number % 200 FROM numbers("number" =
"10000")` into a table partitioned by `p` | fails: `Too many open
partitions 128` | succeeds: 10,000 rows in 200 partitions |
| Data files per partition after one INSERT | up to one per writer | one
per partition (200 identity partitions: 200 files; 300 `day(ts)`
partitions: 300 files) |
| Sink exchange in `EXPLAIN SHAPE PLAN` |
`DistributionSpecHiveTableSinkUnPartitioned` (random) |
`DistributionSpecExternalTableSinkHashPartitioned` |
**How this PR fixes it.**
- **Iceberg connector.** `IcebergWritePlanProvider.getWriteDistribution`
returns an external hash distribution. The route columns are the
partition fields' source columns, the partition function is
`iceberg_transform` with the transforms in spec order (for example
`day,identity,bucket[4]`), and the writer assignment is `SKEWED`, so a
large partition can still scale out. It reads the statement's pinned
write schema when one is active. It returns `null` (today's random
parallel write) for an unpartitioned table, a partition field on a
nested source (route columns must be top-level sink outputs), a
partition field whose transformed value is BINARY or FIXED (`identity`
or `truncate` of a binary or fixed column: the BE holds that value in a
`ColumnVarbinary`, which has no shuffle hash; `bucket` of one yields an
INT and is routed), or a backend older than execution version 18.
- **BE.** The external partition function factory registers
`iceberg_transform`. It reuses `IcebergInsertPartitionFunction`, the
MERGE insert-side router, which applies the same
`PartitionColumnTransforms` as the Iceberg writer. A transform the
writer cannot apply fails the partitioner the same way it fails the
writer; it never falls back to hashing the raw value.
- **Mixed versions.** Older backends reject the new function name, so BE
execution version 18 (17 is taken on master by the `map_agg_v2` frame
change) marks the capability and FE `max_be_exec_version` moves to 18.
The connector reads the version at optimize time through a new
`ConnectorSession#getBeExecVersion()` (default `Integer.MAX_VALUE`). The
engine session returns the version the planner pinned for the statement
(#68837), which is also what the backends receive in the query options
and what `ConnectorWriteHandle#getBeExecVersion()` carries into
`planWrite`, so changing `be_exec_version` while a statement is planned
cannot put the new function into a plan sent at an older version.
Outside a planned statement it returns `Config.be_exec_version`. With
`be_exec_version` below 18 the plan stays as today: the sink keeps the
random distribution, and the INSERT above still fails with `Too many
open partitions 128`. branch-4.x falls back to a single gathered writer
instead.
- **The cap is now per writer's share.** Each writer receives the
partitions whose hash lands on it, so
`table_sink_partition_write_max_partition_nums_per_writer` limits that
share rather than the whole INSERT. One INSERT can span roughly writers
× 128 partitions, less the hash imbalance; a writer whose share goes
over the cap still fails. branch-4.x routes the same way. On a local
single-BE cluster the 200-partition INSERT fails with one writer and the
300-partition `day(ts)` INSERT fails with one or two, and both pass with
three or more (`parallel_pipeline_task_num`).
- The SPI addition is a default method; the plugin-surface baseline is
refreshed. The SPI is unreleased, so `connector.plugin.api.version` is
unchanged. No Thrift change: the function name and its options travel in
the existing `TExternalTableSinkHashPartitionInfo` fields.
Original author of the branch-4.x change: @suxiaogang223
**Results.** On a local single-BE cluster with the Iceberg REST docker
environment, the 200-partition INSERT that fails on master succeeds, a
300-partition `day(ts)` INSERT succeeds too, and every partition holds
exactly one data file. Setting `be_exec_version = 16` brings back the
random distribution and the original error, which is the mixed-version
fallback.
Co-authored-by: Socrates <[email protected]>
---
be/src/agent/be_exec_version_manager.cpp | 4 +-
be/src/agent/be_exec_version_manager.h | 1 +
.../external_partition_function_factory.cpp | 57 +++++++++
.../external_partition_function_factory_test.cpp | 26 ++++
.../external_table_sink_hash_partitioner_test.cpp | 37 ++++++
.../main/java/org/apache/doris/common/Config.java | 4 +-
.../iceberg/IcebergWritePlanProvider.java | 65 ++++++++++
.../iceberg/IcebergWritePlanProviderTest.java | 135 +++++++++++++++++++++
.../doris/connector/spi/ConnectorSession.java | 11 ++
.../test/resources/connector-plugin-surface.txt | 1 +
.../doris/connector/ConnectorSessionBuilder.java | 20 ++-
.../doris/connector/ConnectorSessionImpl.java | 17 ++-
.../doris/connector/ConnectorSessionImplTest.java | 42 +++++++
.../write/test_iceberg_write_binary_partitions.out | 24 ++++
.../test_iceberg_write_partition_distribution.out | 25 ++++
.../test_iceberg_write_binary_partitions.groovy | 20 +++
...est_iceberg_write_partition_distribution.groovy | 128 +++++++++++++++++++
17 files changed, 612 insertions(+), 5 deletions(-)
diff --git a/be/src/agent/be_exec_version_manager.cpp
b/be/src/agent/be_exec_version_manager.cpp
index 1ccdf978688..960159e8d15 100644
--- a/be/src/agent/be_exec_version_manager.cpp
+++ b/be/src/agent/be_exec_version_manager.cpp
@@ -156,8 +156,10 @@ void
BeExecVersionManager::check_function_compatibility(int current_be_exec_vers
// a. support Paimon table writes.
// 17: start from master
// a. use exact logical frames for map_agg_v2 aggregate state serialization.
+// 18: start from master
+// a. support Iceberg partition transform routing in the external sink
exchange.
-const int BeExecVersionManager::max_be_exec_version =
SUPPORT_MAP_AGG_V2_EXACT_FRAME_VERSION;
+const int BeExecVersionManager::max_be_exec_version =
SUPPORT_ICEBERG_TRANSFORM_ROUTING_VERSION;
const int BeExecVersionManager::min_be_exec_version = 0;
std::map<std::string, std::set<int>>
BeExecVersionManager::_function_change_map {};
std::map<std::string, std::set<int>>
BeExecVersionManager::_function_alternative_map {};
diff --git a/be/src/agent/be_exec_version_manager.h
b/be/src/agent/be_exec_version_manager.h
index 01e5f442bad..53564005f0e 100644
--- a/be/src/agent/be_exec_version_manager.h
+++ b/be/src/agent/be_exec_version_manager.h
@@ -39,6 +39,7 @@ constexpr inline int SUPPORT_TIMESTAMP_NS_VERSION = 14;
constexpr inline int SUPPORT_HIVE_OPEN_CSV_VERSION = 15;
constexpr inline int SUPPORT_PAIMON_WRITE_VERSION = 16;
constexpr inline int SUPPORT_MAP_AGG_V2_EXACT_FRAME_VERSION = 17;
+constexpr inline int SUPPORT_ICEBERG_TRANSFORM_ROUTING_VERSION = 18;
class BeExecVersionManager {
public:
diff --git
a/be/src/exec/partitioner/external/external_partition_function_factory.cpp
b/be/src/exec/partitioner/external/external_partition_function_factory.cpp
index 5b312900762..9cd19ae6c33 100644
--- a/be/src/exec/partitioner/external/external_partition_function_factory.cpp
+++ b/be/src/exec/partitioner/external/external_partition_function_factory.cpp
@@ -17,6 +17,7 @@
#include "exec/partitioner/external/external_partition_function_factory.h"
#include <charconv>
+#include <string>
#include <string_view>
#include "common/status.h"
@@ -89,6 +90,58 @@ Status create_paimon_fixed_bucket_function(
return Status::OK();
}
+Status parse_transforms(std::string_view value, std::vector<std::string>*
const transforms) {
+ while (true) {
+ size_t separator = value.find(',');
+ std::string_view item = value.substr(0, separator);
+ if (item.empty()) {
+ return Status::InvalidArgument("Empty Iceberg partition transform
in '{}'", value);
+ }
+ transforms->emplace_back(item);
+ if (separator == std::string_view::npos) {
+ return Status::OK();
+ }
+ value.remove_prefix(separator + 1);
+ }
+}
+
+// Routes each row by its Iceberg partition value: the hash covers the
transformed source columns,
+// so rows of one partition reach one logical partition no matter how the raw
values differ.
+Status create_iceberg_transform_function(
+ const TExternalTableSinkHashPartitionInfo& partition_info,
+ PartitionerBase::HashValType logical_partition_count,
ShuffleHashMethod hash_method,
+ const std::vector<TExpr>& partition_exprs,
+ std::unique_ptr<PartitionFunction>* const partition_function) {
+ if (!partition_info.__isset.partition_function_options) {
+ return Status::InvalidArgument("Iceberg transform partition function
requires options");
+ }
+ const auto& options = partition_info.partition_function_options;
+ if (options.size() != 1 || !options.contains("transforms")) {
+ return Status::InvalidArgument(
+ "Iceberg transform partition function requires exactly the
transforms option");
+ }
+ std::vector<std::string> transforms;
+ RETURN_IF_ERROR(parse_transforms(options.at("transforms"), &transforms));
+ if (transforms.size() != partition_exprs.size()) {
+ return Status::InvalidArgument(
+ "Iceberg partition transform count {} does not match partition
expression count {}",
+ transforms.size(), partition_exprs.size());
+ }
+ std::vector<TIcebergPartitionField> fields;
+ fields.reserve(partition_exprs.size());
+ for (size_t index = 0; index < partition_exprs.size(); ++index) {
+ TIcebergPartitionField field;
+ field.__set_transform(transforms[index]);
+ field.__set_source_expr(partition_exprs[index]);
+ fields.emplace_back(std::move(field));
+ }
+ auto function = std::make_unique<IcebergInsertPartitionFunction>(
+ logical_partition_count, hash_method, std::vector<TExpr> {},
std::move(fields));
+ RETURN_IF_ERROR(function->init({}));
+ *partition_function = std::move(function);
+ return Status::OK();
+}
+
Status create_legacy_external_partition_function(
const TExternalTableSinkHashPartitionInfo& partition_info,
PartitionerBase::HashValType logical_partition_count,
ShuffleHashMethod hash_method,
@@ -173,6 +226,10 @@ Status create_external_partition_function(
return create_paimon_fixed_bucket_function(partition_info,
logical_partition_count,
partition_exprs,
partition_function);
}
+ if (partition_info.partition_function == "iceberg_transform") {
+ return create_iceberg_transform_function(partition_info,
logical_partition_count,
+ hash_method, partition_exprs,
partition_function);
+ }
if (partition_info.partition_function != "direct_hash") {
return Status::NotSupported("Unsupported external sink partition
function '{}'",
partition_info.partition_function);
diff --git
a/be/test/exec/partitioner/external_partition_function_factory_test.cpp
b/be/test/exec/partitioner/external_partition_function_factory_test.cpp
index e03265f6830..de0bd75b92b 100644
--- a/be/test/exec/partitioner/external_partition_function_factory_test.cpp
+++ b/be/test/exec/partitioner/external_partition_function_factory_test.cpp
@@ -129,4 +129,30 @@ TEST(ExternalPartitionFunctionFactoryTest,
ReadsBranch41PaimonFixedBucketEncodin
EXPECT_EQ(function->partition_count(), 3);
}
+TEST(ExternalPartitionFunctionFactoryTest, IcebergTransformValidatesOptions) {
+ auto create = [](const std::map<std::string, std::string>* options) {
+ TExternalTableSinkHashPartitionInfo info;
+ info.__set_partition_function("iceberg_transform");
+ if (options != nullptr) {
+ info.__set_partition_function_options(*options);
+ }
+
info.__set_writer_assignment(TExternalTableSinkWriterAssignment::SKEWED);
+ std::unique_ptr<PartitionFunction> function;
+ return create_external_partition_function(info, 3,
ShuffleHashMethod::CRC32, {}, &function);
+ };
+
+ EXPECT_NE(create(nullptr).to_string().find("requires options"),
std::string::npos);
+ const std::map<std::string, std::string> extra_option {{"transforms",
"identity"},
+ {"unexpected",
"value"}};
+ EXPECT_NE(create(&extra_option).to_string().find("requires exactly the
transforms option"),
+ std::string::npos);
+ const std::map<std::string, std::string> empty_transform {{"transforms",
"identity,,day"}};
+ EXPECT_NE(create(&empty_transform).to_string().find("Empty Iceberg
partition transform"),
+ std::string::npos);
+ // One transform per route expression; this call routes no expressions.
+ const std::map<std::string, std::string> one_transform {{"transforms",
"bucket[16]"}};
+ EXPECT_NE(create(&one_transform).to_string().find("does not match
partition expression count"),
+ std::string::npos);
+}
+
} // namespace doris
diff --git
a/be/test/exec/partitioner/external_table_sink_hash_partitioner_test.cpp
b/be/test/exec/partitioner/external_table_sink_hash_partitioner_test.cpp
index 6b53ab4dd16..c28e9a31eda 100644
--- a/be/test/exec/partitioner/external_table_sink_hash_partitioner_test.cpp
+++ b/be/test/exec/partitioner/external_table_sink_hash_partitioner_test.cpp
@@ -199,6 +199,43 @@ TEST_F(ExternalTableSinkHashPartitionerTest,
PaimonFixedBucketIncludesPartitionH
ASSERT_TRUE(partitioner.close(&_state).ok());
}
+TEST_F(ExternalTableSinkHashPartitionerTest,
IcebergTransformHashesTransformedValue) {
+ TExternalTableSinkHashPartitionInfo info;
+ info.__set_partition_function("iceberg_transform");
+ info.__set_partition_function_options({{"transforms", "truncate[10]"}});
+ info.__set_writer_assignment(TExternalTableSinkWriterAssignment::SKEWED);
+ ExternalTableSinkHashPartitioner partitioner(64, ShuffleHashMethod::CRC32,
info);
+ ASSERT_TRUE(partitioner.init({slot_ref()}).ok());
+ ASSERT_TRUE(partitioner.prepare(&_state, *_row_descriptor).ok());
+ ASSERT_TRUE(partitioner.open(&_state).ok());
+
+ // truncate[10] maps 11 and 19 to partition 10, and 20 and 29 to partition
20.
+ Block input = block({11, 19, 20, 29});
+ ASSERT_TRUE(partitioner.do_partitioning(&_state, &input).ok());
+ const auto& channels = partitioner.get_channel_ids();
+ ASSERT_EQ(4, channels.size());
+ EXPECT_EQ(channels[0], channels[1]);
+ EXPECT_EQ(channels[2], channels[3]);
+ EXPECT_EQ(1, input.columns());
+
+ ASSERT_TRUE(partitioner.close(&_state).ok());
+}
+
+TEST_F(ExternalTableSinkHashPartitionerTest,
IcebergUnsupportedTransformFailsClosed) {
+ TExternalTableSinkHashPartitionInfo info;
+ info.__set_partition_function("iceberg_transform");
+ info.__set_partition_function_options({{"transforms", "unsupported"}});
+ info.__set_writer_assignment(TExternalTableSinkWriterAssignment::IDENTITY);
+ ExternalTableSinkHashPartitioner partitioner(4, ShuffleHashMethod::CRC32,
info);
+ ASSERT_TRUE(partitioner.init({slot_ref()}).ok());
+ ASSERT_TRUE(partitioner.prepare(&_state, *_row_descriptor).ok());
+
+ Status status = partitioner.open(&_state);
+ ASSERT_FALSE(status.ok());
+ EXPECT_NE(status.to_string().find("Unsupported Iceberg partition
transform"),
+ std::string::npos);
+}
+
TEST_F(ExternalTableSinkHashPartitionerTest,
PaimonFixedBucketRejectsMissingMetadata) {
TExternalTableSinkHashPartitionInfo info;
info.__set_partition_function("paimon_fixed_bucket");
diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
index 076723aa8ac..7a2df0152c4 100644
--- a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
+++ b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
@@ -2028,9 +2028,11 @@ public class Config extends ConfigBase {
= ALIGNED_EXTERNAL_TABLE_SINK_THRIFT_MIN_BE_EXEC_VERSION;
// Older backends serialize map_agg_v2 keys with the legacy padded frame
format.
public static final int MAP_AGG_V2_EXACT_FRAME_MIN_BE_EXEC_VERSION = 17;
+ // Older backends reject the iceberg_transform external sink partition
function.
+ public static final int ICEBERG_TRANSFORM_ROUTING_MIN_BE_EXEC_VERSION = 18;
@ConfField(mutable = false)
- public static int max_be_exec_version =
MAP_AGG_V2_EXACT_FRAME_MIN_BE_EXEC_VERSION;
+ public static int max_be_exec_version =
ICEBERG_TRANSFORM_ROUTING_MIN_BE_EXEC_VERSION;
/**
* Min data version of backends serialize block.
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java
index 63945f50e35..5a06993830c 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java
@@ -31,6 +31,7 @@ import
org.apache.doris.connector.spi.handle.ConnectorWriteHandle;
import org.apache.doris.connector.spi.handle.WriteOperation;
import org.apache.doris.connector.spi.write.ConnectorRowChangeStyle;
import org.apache.doris.connector.spi.write.ConnectorSinkPlan;
+import org.apache.doris.connector.spi.write.ConnectorWriteDistribution;
import org.apache.doris.connector.spi.write.ConnectorWritePartitionField;
import org.apache.doris.connector.spi.write.ConnectorWritePartitionSpec;
import org.apache.doris.connector.spi.write.ConnectorWritePlanProvider;
@@ -111,6 +112,9 @@ public class IcebergWritePlanProvider implements
ConnectorWritePlanProvider {
}
private static final int SUPPORT_NESTED_PARTITION_WRITE_EXEC_VERSION = 12;
+ // Older backends reject the iceberg_transform external sink partition
function.
+ private static final int SUPPORT_TRANSFORM_ROUTING_EXEC_VERSION = 18;
+ private static final String TRANSFORM_PARTITION_FUNCTION =
"iceberg_transform";
// Legacy IcebergUtils compression-codec property keys (connector-local
copies; iceberg SDK has no
// constant for the doris/spark-sql forms).
@@ -614,6 +618,67 @@ public class IcebergWritePlanProvider implements
ConnectorWritePlanProvider {
return new ConnectorWritePartitionSpec(spec.specId(), fields);
}
+ /**
+ * Routes a partitioned INSERT / INSERT OVERWRITE by the Iceberg partition
value: the BE
+ * {@code iceberg_transform} partition function hashes each partition
field's transformed source column, so the
+ * rows of one partition reach one writer (SKEWED lets a large partition
scale out). Without it every writer
+ * receives every partition, opens a file per partition it sees, and fails
past
+ * {@code table_sink_partition_write_max_partition_nums_per_writer} open
partitions.
+ *
+ * <p>{@code null} keeps the engine's random parallel write: an
unpartitioned table, a partition field on a nested
+ * source (the route columns are top-level sink outputs), a partition
value the backend cannot hash (see the
+ * loop), or a backend older than the routing function.</p>
+ */
+ @Override
+ public ConnectorWriteDistribution getWriteDistribution(ConnectorSession
session,
+ ConnectorTableHandle tableHandle) {
+ if (session.getBeExecVersion() <
SUPPORT_TRANSFORM_ROUTING_EXEC_VERSION) {
+ return null;
+ }
+ IcebergTableHandle handle = (IcebergTableHandle) tableHandle;
+ Optional<IcebergWriteSchemaContext> active =
+ IcebergStatementScope.activeWriteSchema(
+ session, handle.getDbName(), handle.getTableName());
+ Table table = active.isPresent() ? null : resolveTable(session,
handle);
+ PartitionSpec spec = active.isPresent()
+ ? active.get().getPartitionSpec() : table.spec();
+ if (spec == null || !spec.isPartitioned()) {
+ return null;
+ }
+ Schema schema = active.isPresent()
+ ? active.get().getSchema() : table.schema();
+ List<String> routeColumns = new ArrayList<>();
+ List<String> transforms = new ArrayList<>();
+ for (PartitionField field : spec.fields()) {
+ String sourceColumnName = findTopLevelColumnName(schema,
field.sourceId());
+ if (sourceColumnName == null) {
+ return null;
+ }
+ // The exchange hashes the transformed value, and the BE column of
a BINARY or FIXED value
+ // (ColumnVarbinary) has no shuffle hash: identity(binary_key) or
truncate(1, binary_key) would fail the
+ // INSERT with "update_crc32c_batch is not supported for
ColumnVarbinary" as soon as it has two writers
+ // to route to. bucket(16, binary_key) yields an INT and is routed.
+ Type.TypeID transformedType =
field.transform().getResultType(schema.findType(field.sourceId())).typeId();
+ if (transformedType == Type.TypeID.BINARY || transformedType ==
Type.TypeID.FIXED) {
+ return null;
+ }
+ routeColumns.add(sourceColumnName);
+ transforms.add(field.transform().toString());
+ }
+ return ConnectorWriteDistribution.externalHash(routeColumns,
TRANSFORM_PARTITION_FUNCTION,
+ Collections.singletonMap("transforms", String.join(",",
transforms)),
+ ConnectorWriteDistribution.WriterAssignment.SKEWED);
+ }
+
+ private static String findTopLevelColumnName(Schema schema, int sourceId) {
+ for (NestedField column : schema.columns()) {
+ if (column.fieldId() == sourceId) {
+ return column.name();
+ }
+ }
+ return null;
+ }
+
private static String findPartitionSourceColumnName(Schema schema, int
sourceId) {
List<NestedField> columns = schema.columns();
for (NestedField column : columns) {
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java
index 37e5734bf6f..b265391bb17 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java
@@ -30,6 +30,7 @@ import org.apache.doris.connector.spi.handle.WriteOperation;
import org.apache.doris.connector.spi.mvcc.ConnectorMvccSnapshot;
import org.apache.doris.connector.spi.write.ConnectorRowChangeStyle;
import org.apache.doris.connector.spi.write.ConnectorSinkPlan;
+import org.apache.doris.connector.spi.write.ConnectorWriteDistribution;
import org.apache.doris.connector.spi.write.ConnectorWritePartitionField;
import org.apache.doris.connector.spi.write.ConnectorWritePartitionSpec;
import org.apache.doris.connector.spi.write.ConnectorWriteSortColumn;
@@ -85,6 +86,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.EnumSet;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -1491,6 +1493,128 @@ public class IcebergWritePlanProviderTest {
Assertions.assertEquals(Integer.valueOf(3), field.getSourceId());
}
+ // ───────────────────────────── getWriteDistribution (partition-hash
INSERT routing) ─────────────────────────────
+ //
+ // WHY: without a declared distribution a partitioned INSERT is spread
randomly, so every writer opens a file for
+ // every partition it sees and the write fails once one writer holds more
than
+ // table_sink_partition_write_max_partition_nums_per_writer (128) open
partitions. The BE iceberg_transform
+ // function hashes the transformed partition value, so one partition
reaches one writer.
+
+ @Test
+ public void getWriteDistributionRoutesByTheIcebergTransform() {
+ Table table = bucketPartitionedTable(freshCatalog());
+ ConnectorWriteDistribution distribution = providerFor(table,
contextWithStorage())
+ .getWriteDistribution(sessionFor(table, contextWithStorage()),
new IcebergTableHandle("db1", "tb"));
+
+ Assertions.assertNotNull(distribution, "a partitioned INSERT must be
routed by its partition value");
+ Assertions.assertEquals(ConnectorWriteDistribution.Mode.EXTERNAL_HASH,
distribution.getMode());
+ // The route column is the source column ("id"), not the partition
field ("id_bucket"): fe-core binds it to
+ // the sink output, and the BE applies the transform.
+ Assertions.assertEquals(Collections.singletonList("id"),
distribution.getRouteColumns());
+ Assertions.assertEquals("iceberg_transform",
distribution.getPartitionFunction());
+ Assertions.assertEquals(Collections.singletonMap("transforms",
"bucket[16]"),
+ distribution.getPartitionFunctionOptions());
+
Assertions.assertEquals(ConnectorWriteDistribution.WriterAssignment.SKEWED,
distribution.getWriterAssignment());
+ }
+
+ @Test
+ public void getWriteDistributionCarriesEveryPartitionFieldInSpecOrder() {
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "name", Types.StringType.get()),
+ Types.NestedField.optional(3, "ts",
Types.TimestampType.withoutZone()));
+ Table table = freshCatalog().createTable(TableIdentifier.of("db1",
"multi_field"), schema,
+
PartitionSpec.builderFor(schema).day("ts").identity("name").bucket("id",
4).build());
+
+ ConnectorWriteDistribution distribution = providerFor(table,
contextWithStorage())
+ .getWriteDistribution(sessionFor(table, contextWithStorage()),
+ new IcebergTableHandle("db1", "multi_field"));
+
+ // The BE pairs the i-th route expression with the i-th transform, so
both lists follow the spec order.
+ Assertions.assertEquals(Arrays.asList("ts", "name", "id"),
distribution.getRouteColumns());
+ Assertions.assertEquals(Collections.singletonMap("transforms",
"day,identity,bucket[4]"),
+ distribution.getPartitionFunctionOptions());
+ }
+
+ @Test
+ public void
getWriteDistributionKeepsRandomWritesWhenRoutingIsUnavailable() {
+ InMemoryCatalog catalog = freshCatalog();
+ partitionedSortedTable(catalog);
+ Table unpartitioned = unpartitionedUnsortedTable(catalog);
+ Assertions.assertNull(providerFor(unpartitioned,
contextWithStorage()).getWriteDistribution(
+ sessionFor(unpartitioned, contextWithStorage()), new
IcebergTableHandle("db1", "t2")));
+
+ // The route columns are top-level sink outputs; a nested source has
none, so keep the random write.
+ // MUTATION: resolving the nested source to its top-level struct
column -> routes by "payload" -> red.
+ Schema nestedSchema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "payload", Types.StructType.of(
+ Types.NestedField.optional(3, "part",
Types.IntegerType.get()))));
+ Table nested = catalog.createTable(TableIdentifier.of("db1",
"nested_partition"), nestedSchema,
+ PartitionSpec.builderFor(nestedSchema).bucket("payload.part",
8).build());
+ Assertions.assertNull(providerFor(nested,
contextWithStorage()).getWriteDistribution(
+ sessionFor(nested, contextWithStorage()), new
IcebergTableHandle("db1", "nested_partition")));
+
+ // A backend older than exec version 18 rejects iceberg_transform, so
the INSERT keeps the random write it
+ // already runs there. MUTATION: dropping the version gate -> red.
+ Table partitioned = bucketPartitionedTable(catalog);
+ IcebergTableHandle handle = new IcebergTableHandle("db1", "tb");
+ Assertions.assertNull(providerFor(partitioned,
contextWithStorage()).getWriteDistribution(
+ sessionFor(partitioned,
contextWithStorage()).beExecVersion(17), handle));
+ Assertions.assertNotNull(providerFor(partitioned,
contextWithStorage()).getWriteDistribution(
+ sessionFor(partitioned,
contextWithStorage()).beExecVersion(18), handle));
+ }
+
+ @Test
+ public void getWriteDistributionUsesTheStatementPinnedSchema() {
+ Table table = bucketPartitionedTable(freshCatalog());
+ IcebergWritePlanProvider provider = providerFor(table,
contextWithStorage());
+ WriteSession session = sessionFor(table, contextWithStorage());
+ IcebergTableHandle handle = new IcebergTableHandle("db1", "tb");
+ // Binding the INSERT pins the write schema for the statement; the
sink output carries its column names.
+ provider.getWriteColumns(session, handle, Optional.empty());
+ table.updateSchema().renameColumn("id", "renamed_id").commit();
+
+ // MUTATION: reading the live table instead of the pinned schema ->
routes by "renamed_id", which the bound
+ // sink output does not have -> red.
+ Assertions.assertEquals(Collections.singletonList("id"),
+ provider.getWriteDistribution(session,
handle).getRouteColumns());
+ }
+
+ @Test
+ public void
getWriteDistributionKeepsRandomWritesForBinaryPartitionValues() {
+ // The exchange hashes the transformed value, and the BE column of a
BINARY or FIXED value has no shuffle
+ // hash: routing identity(binary_key) fails the INSERT once it has two
writers. bucket() turns the bytes
+ // into an INT, so that one is routed. MUTATION: dropping the
transformed-type check -> the first four
+ // tables route -> red.
+ InMemoryCatalog catalog = freshCatalog();
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "binary_key",
Types.BinaryType.get()),
+ Types.NestedField.optional(3, "fixed_key",
Types.FixedType.ofLength(4)));
+ Map<String, PartitionSpec> unroutable = new LinkedHashMap<>();
+ unroutable.put("identity_binary",
PartitionSpec.builderFor(schema).identity("binary_key").build());
+ unroutable.put("truncate_binary",
PartitionSpec.builderFor(schema).truncate("binary_key", 1).build());
+ unroutable.put("identity_fixed",
PartitionSpec.builderFor(schema).identity("fixed_key").build());
+ // One unhashable field is enough: the route hashes every field of the
spec together.
+ unroutable.put("int_and_binary",
+
PartitionSpec.builderFor(schema).identity("id").identity("binary_key").build());
+ for (Map.Entry<String, PartitionSpec> entry : unroutable.entrySet()) {
+ Table table = catalog.createTable(TableIdentifier.of("db1",
entry.getKey()), schema, entry.getValue());
+ Assertions.assertNull(providerFor(table,
contextWithStorage()).getWriteDistribution(
+ sessionFor(table, contextWithStorage()), new
IcebergTableHandle("db1", entry.getKey())),
+ entry.getKey());
+ }
+
+ Table bucketed = catalog.createTable(TableIdentifier.of("db1",
"bucket_binary"), schema,
+ PartitionSpec.builderFor(schema).bucket("binary_key",
16).build());
+ ConnectorWriteDistribution distribution = providerFor(bucketed,
contextWithStorage()).getWriteDistribution(
+ sessionFor(bucketed, contextWithStorage()), new
IcebergTableHandle("db1", "bucket_binary"));
+ Assertions.assertEquals(Collections.singletonList("binary_key"),
distribution.getRouteColumns());
+ Assertions.assertEquals(Collections.singletonMap("transforms",
"bucket[16]"),
+ distribution.getPartitionFunctionOptions());
+ }
+
@Test
public void oldBeRejectsDataWritesUsingNestedPartitionSources() {
InMemoryCatalog catalog = freshCatalog();
@@ -2194,6 +2318,7 @@ public class IcebergWritePlanProviderTest {
// Default to a live memoizing scope so a v3 row-level DML does not
trip planWrite's NONE fail-loud; the
// dedicated fail-loud test forces NONE via noScope().
private ConnectorStatementScope statementScope = new
TestStatementScope();
+ private int beExecVersion = Integer.MAX_VALUE;
WriteSession(ConnectorTransaction txn) {
this(txn, "q");
@@ -2204,6 +2329,16 @@ public class IcebergWritePlanProviderTest {
this.queryId = queryId;
}
+ WriteSession beExecVersion(int version) {
+ this.beExecVersion = version;
+ return this;
+ }
+
+ @Override
+ public int getBeExecVersion() {
+ return beExecVersion;
+ }
+
/** Shares an explicit scope (so a test can pre-fill the
rewritable-delete supply the write then reads). */
WriteSession withScope(ConnectorStatementScope scope) {
this.statementScope = scope;
diff --git
a/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorSession.java
b/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorSession.java
index 111e46ac7ff..e305338266e 100644
---
a/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorSession.java
+++
b/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorSession.java
@@ -114,6 +114,17 @@ public interface ConnectorSession {
getSessionProperties().get("enable_external_scan_task_reuse"));
}
+ /**
+ * Query-wide backend execution version selected by the engine, for
planning decisions that depend on a
+ * backend capability (for example a write distribution whose partition
function older backends reject).
+ * Mirrors {@link
org.apache.doris.connector.spi.handle.ConnectorWriteHandle#getBeExecVersion()},
which
+ * carries the same value at {@code planWrite}. Offline and third-party
sessions default to the
+ * newest-capability side of any fence.
+ */
+ default int getBeExecVersion() {
+ return Integer.MAX_VALUE;
+ }
+
/**
* Returns the transaction this session is currently bound to, if any.
*
diff --git
a/fe/fe-connector/fe-connector-spi/src/test/resources/connector-plugin-surface.txt
b/fe/fe-connector/fe-connector-spi/src/test/resources/connector-plugin-surface.txt
index 134d018f5c2..35ee4c05eb6 100644
---
a/fe/fe-connector/fe-connector-spi/src/test/resources/connector-plugin-surface.txt
+++
b/fe/fe-connector/fe-connector-spi/src/test/resources/connector-plugin-surface.txt
@@ -87,6 +87,7 @@
org.apache.doris.connector.spi.ConnectorProvider#validateCreateTable(java.util.M
org.apache.doris.connector.spi.ConnectorProvider#validateProperties(java.util.Map):void
org.apache.doris.connector.spi.ConnectorProvider#validatePropertiesForUpdate(java.util.Map,java.util.Map):void
org.apache.doris.connector.spi.ConnectorSession#allocateTransactionId():long
+org.apache.doris.connector.spi.ConnectorSession#getBeExecVersion():int
org.apache.doris.connector.spi.ConnectorSession#getCatalogId():long
org.apache.doris.connector.spi.ConnectorSession#getCatalogName():java.lang.String
org.apache.doris.connector.spi.ConnectorSession#getCatalogProperties():java.util.Map
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/connector/ConnectorSessionBuilder.java
b/fe/fe-core/src/main/java/org/apache/doris/connector/ConnectorSessionBuilder.java
index bdaeae2f4e9..d5989123614 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/connector/ConnectorSessionBuilder.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/connector/ConnectorSessionBuilder.java
@@ -176,7 +176,25 @@ public final class ConnectorSessionBuilder {
}
return new ConnectorSessionImpl(queryId, user, timeZone, locale,
catalogId, catalogName, catalogProperties, sessionProperties,
sid, cred,
- captureStatementScope());
+ captureStatementScope(), captureBeExecVersion());
+ }
+
+ /**
+ * Captures the backend execution version that NereidsPlanner pinned for
the current statement, which the
+ * query options sent to the backends and the write handle carry too. It
is read off the same statement
+ * context as {@link #captureStatementScope()}. Without a statement
context, or before the planner pins the
+ * version (offline planning, commands that plan no query, tests), it is
{@code Config.be_exec_version}.
+ */
+ private int captureBeExecVersion() {
+ ConnectContext ctx = connectContext != null ? connectContext :
ConnectContext.get();
+ if (ctx == null) {
+ return Config.be_exec_version;
+ }
+ StatementContext sc = ctx.getStatementContext();
+ if (sc == null || !sc.hasBeExecVersion()) {
+ return Config.be_exec_version;
+ }
+ return sc.getBeExecVersion();
}
/**
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/connector/ConnectorSessionImpl.java
b/fe/fe-core/src/main/java/org/apache/doris/connector/ConnectorSessionImpl.java
index 9a2624c5e81..ffd66d2b0cd 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/connector/ConnectorSessionImpl.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/connector/ConnectorSessionImpl.java
@@ -18,6 +18,7 @@
package org.apache.doris.connector;
import org.apache.doris.catalog.Env;
+import org.apache.doris.common.Config;
import org.apache.doris.connector.spi.ConnectorDelegatedCredential;
import org.apache.doris.connector.spi.ConnectorSession;
import org.apache.doris.connector.spi.ConnectorStatementScope;
@@ -53,6 +54,9 @@ public class ConnectorSessionImpl implements ConnectorSession
{
// reachable) so off-thread scan pumps that reuse this one session still
reach it. NONE when there is no
// live statement context (offline planning, tests) -- then
getStatementScope() memoizes nothing.
private final ConnectorStatementScope statementScope;
+ // The backend execution version the planner pinned for the statement,
captured at construction like the
+ // statement scope; Config.be_exec_version when the session is built
outside a planned statement.
+ private final int beExecVersion;
// Otherwise-immutable session; this is bound once by the insert executor
at write time
// for connectors using the SPI transaction model (e.g. maxcompute), and
read back by the
// connector's planWrite via getCurrentTransaction(). volatile for
cross-thread visibility.
@@ -62,13 +66,14 @@ public class ConnectorSessionImpl implements
ConnectorSession {
long catalogId, String catalogName, Map<String, String>
catalogProperties,
Map<String, String> sessionProperties) {
this(queryId, user, timeZone, locale, catalogId, catalogName,
catalogProperties, sessionProperties,
- null, null, ConnectorStatementScope.NONE);
+ null, null, ConnectorStatementScope.NONE,
Config.be_exec_version);
}
ConnectorSessionImpl(String queryId, String user, String timeZone, String
locale,
long catalogId, String catalogName, Map<String, String>
catalogProperties,
Map<String, String> sessionProperties, String sessionId,
- ConnectorDelegatedCredential delegatedCredential,
ConnectorStatementScope statementScope) {
+ ConnectorDelegatedCredential delegatedCredential,
ConnectorStatementScope statementScope,
+ int beExecVersion) {
this.queryId = queryId != null ? queryId : "";
this.user = user != null ? user : "";
this.timeZone = timeZone != null ? timeZone : "UTC";
@@ -82,6 +87,7 @@ public class ConnectorSessionImpl implements ConnectorSession
{
this.sessionId = sessionId;
this.delegatedCredential = delegatedCredential;
this.statementScope = statementScope != null ? statementScope :
ConnectorStatementScope.NONE;
+ this.beExecVersion = beExecVersion;
}
@Override
@@ -165,6 +171,13 @@ public class ConnectorSessionImpl implements
ConnectorSession {
return sessionProperties;
}
+ @Override
+ public int getBeExecVersion() {
+ // The statement's pinned version, which the write handle reports and
the backends receive too, so
+ // optimizer-time choices match planWrite and the dispatched plan.
+ return beExecVersion;
+ }
+
@Override
public long allocateTransactionId() {
return Env.getCurrentEnv().getNextId();
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/connector/ConnectorSessionImplTest.java
b/fe/fe-core/src/test/java/org/apache/doris/connector/ConnectorSessionImplTest.java
index eb2b2b46ce9..d59e5edfc21 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/connector/ConnectorSessionImplTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/connector/ConnectorSessionImplTest.java
@@ -17,6 +17,7 @@
package org.apache.doris.connector;
+import org.apache.doris.common.Config;
import org.apache.doris.connector.spi.ConnectorDelegatedCredential;
import org.apache.doris.connector.spi.ConnectorSession;
import org.apache.doris.connector.spi.ConnectorStatementScope;
@@ -99,6 +100,47 @@ public class ConnectorSessionImplTest {
}
}
+ @Test
+ public void testBeExecVersionIsTheVersionPinnedForTheStatement() {
+ // Connectors gate optimizer-time choices on it (an Iceberg
partition-hash write needs exec version 18). The
+ // planner pins one version per statement, and the query options the
backends receive and the write handle
+ // carry it, so the session must report it too: an ADMIN change of
be_exec_version while the statement is
+ // planned must not put a version-18 partition function into a plan
sent at version 17. MUTATION: reading
+ // Config.be_exec_version in getBeExecVersion() -> red.
+ int original = Config.be_exec_version;
+ ConnectContext ctx = new ConnectContext();
+ ctx.setThreadLocalInfo();
+ try {
+ StatementContext stmtCtx = new StatementContext();
+ ctx.setStatementContext(stmtCtx);
+ stmtCtx.initializeBeExecVersion(17);
+ Config.be_exec_version = 18;
+ Assertions.assertEquals(17,
ConnectorSessionBuilder.from(ctx).build().getBeExecVersion());
+ Assertions.assertEquals(17,
ConnectorSessionBuilder.create().build().getBeExecVersion(),
+ "a session built on the statement's thread reads the
thread-local statement context");
+ } finally {
+ Config.be_exec_version = original;
+ ConnectContext.remove();
+ }
+ }
+
+ @Test
+ public void
testBeExecVersionFallsBackToTheEngineConfigOutsideAPlannedStatement() {
+ // Sessions built where no statement version is pinned (background
loaders, commands that plan no query,
+ // tests) report the configured version.
+ int original = Config.be_exec_version;
+ try {
+ Config.be_exec_version = 16;
+ Assertions.assertEquals(16,
ConnectorSessionBuilder.create().build().getBeExecVersion());
+ ConnectContext ctx = new ConnectContext();
+ ctx.setStatementContext(new StatementContext());
+ Assertions.assertEquals(16,
ConnectorSessionBuilder.from(ctx).build().getBeExecVersion(),
+ "a statement whose version is not pinned yet");
+ } finally {
+ Config.be_exec_version = original;
+ }
+ }
+
@Test
public void testSessionPropertyOverridesCatalogProperty() {
Map<String, String> catalogProps = new HashMap<>();
diff --git
a/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_binary_partitions.out
b/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_binary_partitions.out
index 3789d7e436a..e60f362fd80 100644
---
a/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_binary_partitions.out
+++
b/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_binary_partitions.out
@@ -275,3 +275,27 @@
-- !binary_orc_truncate_INSERT_OVERWRITE_TABLE_null --
7
+-- !binary_multi_writer_identity_INSERT_INTO --
+3130 2500 0 9996
+3131 2500 1 9997
+3132 2500 2 9998
+3133 2500 3 9999
+
+-- !binary_multi_writer_identity_INSERT_OVERWRITE_TABLE --
+3130 2500 0 9996
+3131 2500 1 9997
+3132 2500 2 9998
+3133 2500 3 9999
+
+-- !binary_multi_writer_truncate_INSERT_INTO --
+3130 2500 0 9996
+3131 2500 1 9997
+3132 2500 2 9998
+3133 2500 3 9999
+
+-- !binary_multi_writer_truncate_INSERT_OVERWRITE_TABLE --
+3130 2500 0 9996
+3131 2500 1 9997
+3132 2500 2 9998
+3133 2500 3 9999
+
diff --git
a/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_partition_distribution.out
b/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_partition_distribution.out
new file mode 100644
index 00000000000..23aa7647ba6
--- /dev/null
+++
b/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_partition_distribution.out
@@ -0,0 +1,25 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !identity_plan --
+PhysicalConnectorTableSink
+--PhysicalDistribute[DistributionSpecExternalTableSinkHashPartitioned]
+----PhysicalProject
+------PhysicalTVFRelation
+
+-- !identity_rows --
+10000 200 0 9999
+
+-- !identity_files --
+200 1
+
+-- !day_plan --
+PhysicalConnectorTableSink
+--PhysicalDistribute[DistributionSpecExternalTableSinkHashPartitioned]
+----PhysicalProject
+------PhysicalTVFRelation
+
+-- !day_rows --
+10000 300 2024-01-01T00:00 2024-10-26T23:00
+
+-- !day_files --
+300 1
+
diff --git
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_binary_partitions.groovy
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_binary_partitions.groovy
index 31a592610c8..79995932ee4 100644
---
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_binary_partitions.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_binary_partitions.groovy
@@ -92,6 +92,26 @@ suite("test_iceberg_write_binary_partitions",
"p0,external,iceberg,external_dock
}
}
}
+
+ // The VALUES writes above have a single-instance source, so their
sink has one writer and its exchange
+ // does not hash. A multi-instance source reaches several writers
through a hashing exchange, which must
+ // not route by a binary partition value: the BE cannot hash it
("update_crc32c_batch is not supported
+ // for ColumnVarbinary"). Pin the writers, since a fuzzy session can
pick one.
+ sql "set parallel_pipeline_task_num = 8"
+ for (String transform : ["identity", "truncate"]) {
+ String table = "binary_multi_writer_${transform}"
+ String expression = transform == "identity" ? "binary_key" :
"truncate(1, binary_key)"
+ sql "DROP TABLE IF EXISTS ${table}"
+ spark_iceberg """CREATE TABLE demo.${dbName}.${table} (id BIGINT,
binary_key BINARY)
+ USING iceberg PARTITIONED BY (${expression})
+ TBLPROPERTIES ('format-version'='2',
'write.format.default'='parquet')"""
+ for (String operation : ["INSERT INTO", "INSERT OVERWRITE TABLE"])
{
+ sql """${operation} ${table}
+ SELECT number, CAST(CAST(number % 4 + 10 AS STRING) AS
VARBINARY) FROM numbers("number" = "10000")"""
+ "order_qt_${table}_${operation.replace(' ', '_')}" """
+ SELECT HEX(binary_key), count(*), min(id), max(id) FROM
${table} GROUP BY HEX(binary_key) ORDER BY 1"""
+ }
+ }
} finally {
sql "SWITCH internal"
}
diff --git
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_partition_distribution.groovy
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_partition_distribution.groovy
new file mode 100644
index 00000000000..6e355f33605
--- /dev/null
+++
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_partition_distribution.groovy
@@ -0,0 +1,128 @@
+// 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.
+
+// A partitioned INSERT is routed by the Iceberg partition value, so one
partition reaches one writer. A write
+// touching more than table_sink_partition_write_max_partition_nums_per_writer
(128) partitions then succeeds,
+// and each partition gets one data file instead of one per writer.
+suite("test_iceberg_write_partition_distribution",
+ "p0,external,iceberg,external_docker,external_docker_iceberg") {
+ String enabled = context.config.otherConfigs.get("enableIcebergTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable iceberg test")
+ return
+ }
+
+ String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String catalogName = "test_iceberg_write_partition_distribution"
+ String dbName = "iceberg_write_partition_distribution_db"
+
+ sql """drop catalog if exists ${catalogName}"""
+ sql """
+ create catalog ${catalogName} properties (
+ "type" = "iceberg",
+ "iceberg.catalog.type" = "rest",
+ "uri" = "http://${externalEnvIp}:${restPort}";,
+ "s3.access_key" = "admin",
+ "s3.secret_key" = "password",
+ "s3.endpoint" = "http://${externalEnvIp}:${minioPort}";,
+ "s3.region" = "us-east-1",
+ "meta.cache.iceberg.table.ttl-second" = "0",
+ "meta.cache.iceberg.schema.ttl-second" = "0"
+ )
+ """
+ sql """switch ${catalogName}"""
+ sql """drop database if exists ${dbName} force"""
+ sql """create database ${dbName}"""
+ sql """use ${dbName}"""
+ // The writers follow the source fragment's parallelism, and the
128-partition cap applies to the partitions
+ // routed to each writer. Pin enough writers for 300 partitions: a fuzzy
session can pick one or two.
+ sql """set parallel_pipeline_task_num = 8"""
+
+ // Identity partitions: 200 partitions in one INSERT.
+ sql """drop table if exists identity_partitions"""
+ sql """
+ create table identity_partitions (
+ id int,
+ p int
+ )
+ partition by list (p) ()
+ properties (
+ "format-version" = "2",
+ "write.format.default" = "parquet"
+ )
+ """
+ qt_identity_plan """
+ explain shape plan
+ insert into identity_partitions
+ select number, number % 200 from numbers("number" = "10000")
+ """
+ sql """
+ insert into identity_partitions
+ select number, number % 200 from numbers("number" = "10000")
+ """
+ qt_identity_rows """
+ select count(*), count(distinct p), min(id), max(id) from
identity_partitions
+ """
+ // One writer owns each partition, so every partition holds exactly one
data file.
+ qt_identity_files """
+ select count(*), max(file_count)
+ from (
+ select `partition`, count(*) as file_count
+ from identity_partitions\$files
+ group by `partition`
+ ) partition_file_counts
+ """
+
+ // Transform partitions: rows of one day carry different timestamps, so
routing by the raw source value would
+ // split a day across writers. The hash must cover day(ts).
+ sql """drop table if exists day_partitions"""
+ sql """
+ create table day_partitions (
+ id int,
+ ts datetime
+ )
+ partition by list (day(ts)) ()
+ properties (
+ "format-version" = "2",
+ "write.format.default" = "parquet"
+ )
+ """
+ qt_day_plan """
+ explain shape plan
+ insert into day_partitions
+ select number, hours_add(cast('2024-01-01 00:00:00' as datetime),
number % 7200)
+ from numbers("number" = "10000")
+ """
+ sql """
+ insert into day_partitions
+ select number, hours_add(cast('2024-01-01 00:00:00' as datetime),
number % 7200)
+ from numbers("number" = "10000")
+ """
+ qt_day_rows """
+ select count(*), count(distinct date(ts)), min(ts), max(ts) from
day_partitions
+ """
+ qt_day_files """
+ select count(*), max(file_count)
+ from (
+ select `partition`, count(*) as file_count
+ from day_partitions\$files
+ group by `partition`
+ ) partition_file_counts
+ """
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
