You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by et...@apache.org on 2023/10/17 16:30:32 UTC
[iceberg] branch 1.4.x updated: Flink: Reverting the default custom partitioner for bucket column (#8848) (#8858)
This is an automated email from the ASF dual-hosted git repository.
etudenhoefner pushed a commit to branch 1.4.x
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/1.4.x by this push:
new 4f999f12d6 Flink: Reverting the default custom partitioner for bucket column (#8848) (#8858)
4f999f12d6 is described below
commit 4f999f12d67547430e5d347f002697c42829374b
Author: Eduard Tudenhoefner <et...@gmail.com>
AuthorDate: Tue Oct 17 18:30:24 2023 +0200
Flink: Reverting the default custom partitioner for bucket column (#8848) (#8858)
---
.../org/apache/iceberg/flink/sink/FlinkSink.java | 8 +-----
.../TestBucketPartitionerFlinkIcebergSink.java | 30 ++++++----------------
.../org/apache/iceberg/flink/sink/FlinkSink.java | 8 +-----
.../TestBucketPartitionerFlinkIcebergSink.java | 30 ++++++----------------
.../org/apache/iceberg/flink/sink/FlinkSink.java | 8 +-----
.../TestBucketPartitionerFlinkIcebergSink.java | 30 ++++++----------------
6 files changed, 27 insertions(+), 87 deletions(-)
diff --git a/flink/v1.15/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java b/flink/v1.15/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
index 5882879925..220353d75a 100644
--- a/flink/v1.15/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
+++ b/flink/v1.15/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
@@ -525,13 +525,7 @@ public class FlinkSink {
+ "and table is unpartitioned");
return input;
} else {
- if (BucketPartitionerUtil.hasOneBucketField(partitionSpec)) {
- return input.partitionCustom(
- new BucketPartitioner(partitionSpec),
- new BucketPartitionKeySelector(partitionSpec, iSchema, flinkRowType));
- } else {
- return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
- }
+ return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
}
} else {
if (partitionSpec.isUnpartitioned()) {
diff --git a/flink/v1.15/flink/src/test/java/org/apache/iceberg/flink/sink/TestBucketPartitionerFlinkIcebergSink.java b/flink/v1.15/flink/src/test/java/org/apache/iceberg/flink/sink/TestBucketPartitionerFlinkIcebergSink.java
index 29a0898a1b..9dae43ce5e 100644
--- a/flink/v1.15/flink/src/test/java/org/apache/iceberg/flink/sink/TestBucketPartitionerFlinkIcebergSink.java
+++ b/flink/v1.15/flink/src/test/java/org/apache/iceberg/flink/sink/TestBucketPartitionerFlinkIcebergSink.java
@@ -114,13 +114,19 @@ public class TestBucketPartitionerFlinkIcebergSink {
new BoundedTestSource<>(
allRows.stream().map(converter::toExternal).toArray(Row[]::new)),
ROW_TYPE_INFO)
- .map(converter::toInternal, FlinkCompatibilityUtil.toTypeInfo(SimpleDataUtil.ROW_TYPE));
+ .map(converter::toInternal, FlinkCompatibilityUtil.toTypeInfo(SimpleDataUtil.ROW_TYPE))
+ .partitionCustom(
+ new BucketPartitioner(table.spec()),
+ new BucketPartitionKeySelector(
+ table.spec(),
+ table.schema(),
+ FlinkSink.toFlinkRowType(table.schema(), SimpleDataUtil.FLINK_SCHEMA)));
FlinkSink.forRowData(dataStream)
.table(table)
.tableLoader(tableLoader)
.writeParallelism(parallelism)
- .distributionMode(DistributionMode.HASH)
+ .distributionMode(DistributionMode.NONE)
.append();
env.execute("Test Iceberg DataStream");
@@ -157,26 +163,6 @@ public class TestBucketPartitionerFlinkIcebergSink {
}
}
- /**
- * Verifies the BucketPartitioner is not used when the PartitionSpec has more than 1 bucket, and
- * that it should fallback to input.keyBy
- */
- @ParameterizedTest
- @EnumSource(value = TableSchemaType.class, names = "TWO_BUCKETS")
- public void testMultipleBucketsFallback(TableSchemaType tableSchemaType) throws Exception {
- setupEnvironment(tableSchemaType);
- List<RowData> rows = generateTestDataRows();
-
- appendRowsToTable(rows);
- TableTestStats stats = extractPartitionResults(tableSchemaType);
-
- Assertions.assertThat(stats.totalRowCount).isEqualTo(rows.size());
- for (int i = 0, j = numBuckets; i < numBuckets; i++, j++) {
- // Only 1 file per bucket will be created when falling back to input.keyBy(...)
- Assertions.assertThat((int) stats.numFilesPerBucket.get(i)).isEqualTo(1);
- }
- }
-
/**
* Generating 16 rows to be sent uniformly to all writers (round-robin across 8 writers -> 4
* buckets)
diff --git a/flink/v1.16/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java b/flink/v1.16/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
index 5882879925..220353d75a 100644
--- a/flink/v1.16/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
+++ b/flink/v1.16/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
@@ -525,13 +525,7 @@ public class FlinkSink {
+ "and table is unpartitioned");
return input;
} else {
- if (BucketPartitionerUtil.hasOneBucketField(partitionSpec)) {
- return input.partitionCustom(
- new BucketPartitioner(partitionSpec),
- new BucketPartitionKeySelector(partitionSpec, iSchema, flinkRowType));
- } else {
- return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
- }
+ return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
}
} else {
if (partitionSpec.isUnpartitioned()) {
diff --git a/flink/v1.16/flink/src/test/java/org/apache/iceberg/flink/sink/TestBucketPartitionerFlinkIcebergSink.java b/flink/v1.16/flink/src/test/java/org/apache/iceberg/flink/sink/TestBucketPartitionerFlinkIcebergSink.java
index 29a0898a1b..9dae43ce5e 100644
--- a/flink/v1.16/flink/src/test/java/org/apache/iceberg/flink/sink/TestBucketPartitionerFlinkIcebergSink.java
+++ b/flink/v1.16/flink/src/test/java/org/apache/iceberg/flink/sink/TestBucketPartitionerFlinkIcebergSink.java
@@ -114,13 +114,19 @@ public class TestBucketPartitionerFlinkIcebergSink {
new BoundedTestSource<>(
allRows.stream().map(converter::toExternal).toArray(Row[]::new)),
ROW_TYPE_INFO)
- .map(converter::toInternal, FlinkCompatibilityUtil.toTypeInfo(SimpleDataUtil.ROW_TYPE));
+ .map(converter::toInternal, FlinkCompatibilityUtil.toTypeInfo(SimpleDataUtil.ROW_TYPE))
+ .partitionCustom(
+ new BucketPartitioner(table.spec()),
+ new BucketPartitionKeySelector(
+ table.spec(),
+ table.schema(),
+ FlinkSink.toFlinkRowType(table.schema(), SimpleDataUtil.FLINK_SCHEMA)));
FlinkSink.forRowData(dataStream)
.table(table)
.tableLoader(tableLoader)
.writeParallelism(parallelism)
- .distributionMode(DistributionMode.HASH)
+ .distributionMode(DistributionMode.NONE)
.append();
env.execute("Test Iceberg DataStream");
@@ -157,26 +163,6 @@ public class TestBucketPartitionerFlinkIcebergSink {
}
}
- /**
- * Verifies the BucketPartitioner is not used when the PartitionSpec has more than 1 bucket, and
- * that it should fallback to input.keyBy
- */
- @ParameterizedTest
- @EnumSource(value = TableSchemaType.class, names = "TWO_BUCKETS")
- public void testMultipleBucketsFallback(TableSchemaType tableSchemaType) throws Exception {
- setupEnvironment(tableSchemaType);
- List<RowData> rows = generateTestDataRows();
-
- appendRowsToTable(rows);
- TableTestStats stats = extractPartitionResults(tableSchemaType);
-
- Assertions.assertThat(stats.totalRowCount).isEqualTo(rows.size());
- for (int i = 0, j = numBuckets; i < numBuckets; i++, j++) {
- // Only 1 file per bucket will be created when falling back to input.keyBy(...)
- Assertions.assertThat((int) stats.numFilesPerBucket.get(i)).isEqualTo(1);
- }
- }
-
/**
* Generating 16 rows to be sent uniformly to all writers (round-robin across 8 writers -> 4
* buckets)
diff --git a/flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java b/flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
index 5882879925..220353d75a 100644
--- a/flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
+++ b/flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
@@ -525,13 +525,7 @@ public class FlinkSink {
+ "and table is unpartitioned");
return input;
} else {
- if (BucketPartitionerUtil.hasOneBucketField(partitionSpec)) {
- return input.partitionCustom(
- new BucketPartitioner(partitionSpec),
- new BucketPartitionKeySelector(partitionSpec, iSchema, flinkRowType));
- } else {
- return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
- }
+ return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
}
} else {
if (partitionSpec.isUnpartitioned()) {
diff --git a/flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/sink/TestBucketPartitionerFlinkIcebergSink.java b/flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/sink/TestBucketPartitionerFlinkIcebergSink.java
index 29a0898a1b..9dae43ce5e 100644
--- a/flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/sink/TestBucketPartitionerFlinkIcebergSink.java
+++ b/flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/sink/TestBucketPartitionerFlinkIcebergSink.java
@@ -114,13 +114,19 @@ public class TestBucketPartitionerFlinkIcebergSink {
new BoundedTestSource<>(
allRows.stream().map(converter::toExternal).toArray(Row[]::new)),
ROW_TYPE_INFO)
- .map(converter::toInternal, FlinkCompatibilityUtil.toTypeInfo(SimpleDataUtil.ROW_TYPE));
+ .map(converter::toInternal, FlinkCompatibilityUtil.toTypeInfo(SimpleDataUtil.ROW_TYPE))
+ .partitionCustom(
+ new BucketPartitioner(table.spec()),
+ new BucketPartitionKeySelector(
+ table.spec(),
+ table.schema(),
+ FlinkSink.toFlinkRowType(table.schema(), SimpleDataUtil.FLINK_SCHEMA)));
FlinkSink.forRowData(dataStream)
.table(table)
.tableLoader(tableLoader)
.writeParallelism(parallelism)
- .distributionMode(DistributionMode.HASH)
+ .distributionMode(DistributionMode.NONE)
.append();
env.execute("Test Iceberg DataStream");
@@ -157,26 +163,6 @@ public class TestBucketPartitionerFlinkIcebergSink {
}
}
- /**
- * Verifies the BucketPartitioner is not used when the PartitionSpec has more than 1 bucket, and
- * that it should fallback to input.keyBy
- */
- @ParameterizedTest
- @EnumSource(value = TableSchemaType.class, names = "TWO_BUCKETS")
- public void testMultipleBucketsFallback(TableSchemaType tableSchemaType) throws Exception {
- setupEnvironment(tableSchemaType);
- List<RowData> rows = generateTestDataRows();
-
- appendRowsToTable(rows);
- TableTestStats stats = extractPartitionResults(tableSchemaType);
-
- Assertions.assertThat(stats.totalRowCount).isEqualTo(rows.size());
- for (int i = 0, j = numBuckets; i < numBuckets; i++, j++) {
- // Only 1 file per bucket will be created when falling back to input.keyBy(...)
- Assertions.assertThat((int) stats.numFilesPerBucket.get(i)).isEqualTo(1);
- }
- }
-
/**
* Generating 16 rows to be sent uniformly to all writers (round-robin across 8 writers -> 4
* buckets)