You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "rodmeneses (via GitHub)" <gi...@apache.org> on 2025/01/23 17:43:48 UTC
[PR] Range distribution iceberg sink [iceberg]
rodmeneses opened a new pull request, #12071:
URL: https://github.com/apache/iceberg/pull/12071
(no comment)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Range distribution iceberg sink [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#issuecomment-2817406414
This pull request has been closed due to lack of activity. This is not a judgement on the merit of the PR in any way. It is just a way of keeping the PR queue manageable. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: port range distribution to v2 iceberg sink [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#discussion_r2103486896
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -645,72 +711,135 @@ private DataStream<RowData> distributeDataStream(DataStream<RowData> input) {
DistributionMode mode = flinkWriteConf.distributionMode();
Schema schema = table.schema();
PartitionSpec spec = table.spec();
+ SortOrder sortOrder = table.sortOrder();
+
LOG.info("Write distribution mode is '{}'", mode.modeName());
switch (mode) {
case NONE:
- if (equalityFieldIds.isEmpty()) {
- return input;
- } else {
- LOG.info("Distribute rows by equality fields, because there are equality fields set");
- return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- }
-
+ return distributeDataStreamByNoneDistributionMode(input, schema);
case HASH:
- if (equalityFieldIds.isEmpty()) {
- if (table.spec().isUnpartitioned()) {
- LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and table is unpartitioned");
- return input;
- } else {
- if (BucketPartitionerUtil.hasOneBucketField(spec)) {
- return input.partitionCustom(
- new BucketPartitioner(spec),
- new BucketPartitionKeySelector(spec, schema, flinkRowType));
- } else {
- return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
- }
- }
- } else {
- if (spec.isUnpartitioned()) {
- LOG.info(
- "Distribute rows by equality fields, because there are equality fields set "
- + "and table is unpartitioned");
- return input.keyBy(
- new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- } else {
- for (PartitionField partitionField : spec.fields()) {
- Preconditions.checkState(
- equalityFieldIds.contains(partitionField.sourceId()),
- "In 'hash' distribution mode with equality fields set, partition field '%s' "
- + "should be included in equality fields: '%s'",
- partitionField,
- equalityFieldColumns);
- }
- return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
- }
- }
-
+ return distributeDataStreamByHashDistributionMode(input, schema, spec);
case RANGE:
- if (equalityFieldIds.isEmpty()) {
- LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and {}=range is not supported yet in flink",
- WRITE_DISTRIBUTION_MODE);
- return input;
- } else {
- LOG.info(
- "Distribute rows by equality fields, because there are equality fields set "
- + "and{}=range is not supported yet in flink",
- WRITE_DISTRIBUTION_MODE);
- return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- }
-
+ return distributeDataStreamByRangeDistributionMode(input, schema, spec, sortOrder);
default:
throw new RuntimeException("Unrecognized " + WRITE_DISTRIBUTION_MODE + ": " + mode);
}
}
+ private DataStream<RowData> distributeDataStreamByNoneDistributionMode(
+ DataStream<RowData> input, Schema iSchema) {
+ if (equalityFieldIds.isEmpty()) {
+ return input;
+ } else {
+ LOG.info("Distribute rows by equality fields, because there are equality fields set");
+ return input.keyBy(new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));
+ }
+ }
+
+ private DataStream<RowData> distributeDataStreamByHashDistributionMode(
+ DataStream<RowData> input, Schema iSchema, PartitionSpec partitionSpec) {
+ if (equalityFieldIds.isEmpty()) {
+ if (partitionSpec.isUnpartitioned()) {
+ LOG.warn(
+ "Fallback to use 'none' distribution mode, because there are no equality fields set "
+ + "and table is unpartitioned");
+ return input;
+ } else {
+ return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
+ }
+ } else {
+ if (partitionSpec.isUnpartitioned()) {
+ LOG.info(
+ "Distribute rows by equality fields, because there are equality fields set "
+ + "and table is unpartitioned");
+ return input.keyBy(new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));
+ } else {
+ for (PartitionField partitionField : partitionSpec.fields()) {
+ Preconditions.checkState(
+ equalityFieldIds.contains(partitionField.sourceId()),
+ "In 'hash' distribution mode with equality fields set, partition field '%s' "
+ + "should be included in equality fields: '%s'",
+ partitionField,
+ equalityFieldColumns);
+ }
+ return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
+ }
+ }
+ }
+
+ private DataStream<RowData> distributeDataStreamByRangeDistributionMode(
+ DataStream<RowData> input,
+ Schema iSchema,
+ PartitionSpec partitionSpec,
+ SortOrder sortOrderParam) {
+
+ int writerParallelism =
Review Comment:
sorry, I wasn't very clear earlier. The v2 sink writer parallelism selection is different than the v1 sink. It doesn't use input parallelism if write parallelism is not configured explicitly.
```
@Override
public Builder writeParallelism(int newWriteParallelism) {
writeOptions.put(
FlinkWriteOptions.WRITE_PARALLELISM.key(), Integer.toString(newWriteParallelism));
return this;
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: port range distribution to v2 iceberg sink [iceberg]
Posted by "rodmeneses (via GitHub)" <gi...@apache.org>.
rodmeneses commented on code in PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#discussion_r2121884952
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -645,72 +711,135 @@ private DataStream<RowData> distributeDataStream(DataStream<RowData> input) {
DistributionMode mode = flinkWriteConf.distributionMode();
Schema schema = table.schema();
PartitionSpec spec = table.spec();
+ SortOrder sortOrder = table.sortOrder();
+
LOG.info("Write distribution mode is '{}'", mode.modeName());
switch (mode) {
case NONE:
- if (equalityFieldIds.isEmpty()) {
- return input;
- } else {
- LOG.info("Distribute rows by equality fields, because there are equality fields set");
- return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- }
-
+ return distributeDataStreamByNoneDistributionMode(input, schema);
case HASH:
- if (equalityFieldIds.isEmpty()) {
- if (table.spec().isUnpartitioned()) {
- LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and table is unpartitioned");
- return input;
- } else {
- if (BucketPartitionerUtil.hasOneBucketField(spec)) {
- return input.partitionCustom(
- new BucketPartitioner(spec),
- new BucketPartitionKeySelector(spec, schema, flinkRowType));
- } else {
- return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
- }
- }
- } else {
- if (spec.isUnpartitioned()) {
- LOG.info(
- "Distribute rows by equality fields, because there are equality fields set "
- + "and table is unpartitioned");
- return input.keyBy(
- new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- } else {
- for (PartitionField partitionField : spec.fields()) {
- Preconditions.checkState(
- equalityFieldIds.contains(partitionField.sourceId()),
- "In 'hash' distribution mode with equality fields set, partition field '%s' "
- + "should be included in equality fields: '%s'",
- partitionField,
- equalityFieldColumns);
- }
- return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
- }
- }
-
+ return distributeDataStreamByHashDistributionMode(input, schema, spec);
case RANGE:
- if (equalityFieldIds.isEmpty()) {
- LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and {}=range is not supported yet in flink",
- WRITE_DISTRIBUTION_MODE);
- return input;
- } else {
- LOG.info(
- "Distribute rows by equality fields, because there are equality fields set "
- + "and{}=range is not supported yet in flink",
- WRITE_DISTRIBUTION_MODE);
- return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- }
-
+ return distributeDataStreamByRangeDistributionMode(input, schema, spec, sortOrder);
default:
throw new RuntimeException("Unrecognized " + WRITE_DISTRIBUTION_MODE + ": " + mode);
}
}
+ private DataStream<RowData> distributeDataStreamByNoneDistributionMode(
+ DataStream<RowData> input, Schema iSchema) {
+ if (equalityFieldIds.isEmpty()) {
+ return input;
+ } else {
+ LOG.info("Distribute rows by equality fields, because there are equality fields set");
+ return input.keyBy(new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));
+ }
+ }
+
+ private DataStream<RowData> distributeDataStreamByHashDistributionMode(
+ DataStream<RowData> input, Schema iSchema, PartitionSpec partitionSpec) {
+ if (equalityFieldIds.isEmpty()) {
+ if (partitionSpec.isUnpartitioned()) {
+ LOG.warn(
+ "Fallback to use 'none' distribution mode, because there are no equality fields set "
+ + "and table is unpartitioned");
+ return input;
+ } else {
+ return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
+ }
+ } else {
+ if (partitionSpec.isUnpartitioned()) {
+ LOG.info(
+ "Distribute rows by equality fields, because there are equality fields set "
+ + "and table is unpartitioned");
+ return input.keyBy(new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));
+ } else {
+ for (PartitionField partitionField : partitionSpec.fields()) {
+ Preconditions.checkState(
+ equalityFieldIds.contains(partitionField.sourceId()),
+ "In 'hash' distribution mode with equality fields set, partition field '%s' "
+ + "should be included in equality fields: '%s'",
+ partitionField,
+ equalityFieldColumns);
+ }
+ return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
+ }
+ }
+ }
+
+ private DataStream<RowData> distributeDataStreamByRangeDistributionMode(
+ DataStream<RowData> input,
+ Schema iSchema,
+ PartitionSpec partitionSpec,
+ SortOrder sortOrderParam) {
+
+ int writerParallelism =
Review Comment:
Hi @stevenzwu I have rebased and this is ready! Thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: port range distribution to v2 iceberg sink [iceberg]
Posted by "rodmeneses (via GitHub)" <gi...@apache.org>.
rodmeneses commented on code in PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#discussion_r2103206341
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -645,72 +711,135 @@ private DataStream<RowData> distributeDataStream(DataStream<RowData> input) {
DistributionMode mode = flinkWriteConf.distributionMode();
Schema schema = table.schema();
PartitionSpec spec = table.spec();
+ SortOrder sortOrder = table.sortOrder();
+
LOG.info("Write distribution mode is '{}'", mode.modeName());
switch (mode) {
case NONE:
- if (equalityFieldIds.isEmpty()) {
- return input;
- } else {
- LOG.info("Distribute rows by equality fields, because there are equality fields set");
- return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- }
-
+ return distributeDataStreamByNoneDistributionMode(input, schema);
case HASH:
- if (equalityFieldIds.isEmpty()) {
- if (table.spec().isUnpartitioned()) {
- LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and table is unpartitioned");
- return input;
- } else {
- if (BucketPartitionerUtil.hasOneBucketField(spec)) {
- return input.partitionCustom(
- new BucketPartitioner(spec),
- new BucketPartitionKeySelector(spec, schema, flinkRowType));
- } else {
- return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
- }
- }
- } else {
- if (spec.isUnpartitioned()) {
- LOG.info(
- "Distribute rows by equality fields, because there are equality fields set "
- + "and table is unpartitioned");
- return input.keyBy(
- new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- } else {
- for (PartitionField partitionField : spec.fields()) {
- Preconditions.checkState(
- equalityFieldIds.contains(partitionField.sourceId()),
- "In 'hash' distribution mode with equality fields set, partition field '%s' "
- + "should be included in equality fields: '%s'",
- partitionField,
- equalityFieldColumns);
- }
- return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
- }
- }
-
+ return distributeDataStreamByHashDistributionMode(input, schema, spec);
case RANGE:
- if (equalityFieldIds.isEmpty()) {
- LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and {}=range is not supported yet in flink",
- WRITE_DISTRIBUTION_MODE);
- return input;
- } else {
- LOG.info(
- "Distribute rows by equality fields, because there are equality fields set "
- + "and{}=range is not supported yet in flink",
- WRITE_DISTRIBUTION_MODE);
- return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- }
-
+ return distributeDataStreamByRangeDistributionMode(input, schema, spec, sortOrder);
default:
throw new RuntimeException("Unrecognized " + WRITE_DISTRIBUTION_MODE + ": " + mode);
}
}
+ private DataStream<RowData> distributeDataStreamByNoneDistributionMode(
+ DataStream<RowData> input, Schema iSchema) {
+ if (equalityFieldIds.isEmpty()) {
+ return input;
+ } else {
+ LOG.info("Distribute rows by equality fields, because there are equality fields set");
+ return input.keyBy(new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));
+ }
+ }
+
+ private DataStream<RowData> distributeDataStreamByHashDistributionMode(
+ DataStream<RowData> input, Schema iSchema, PartitionSpec partitionSpec) {
+ if (equalityFieldIds.isEmpty()) {
+ if (partitionSpec.isUnpartitioned()) {
+ LOG.warn(
+ "Fallback to use 'none' distribution mode, because there are no equality fields set "
+ + "and table is unpartitioned");
+ return input;
+ } else {
+ return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
+ }
+ } else {
+ if (partitionSpec.isUnpartitioned()) {
+ LOG.info(
+ "Distribute rows by equality fields, because there are equality fields set "
+ + "and table is unpartitioned");
+ return input.keyBy(new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));
+ } else {
+ for (PartitionField partitionField : partitionSpec.fields()) {
+ Preconditions.checkState(
+ equalityFieldIds.contains(partitionField.sourceId()),
+ "In 'hash' distribution mode with equality fields set, partition field '%s' "
+ + "should be included in equality fields: '%s'",
+ partitionField,
+ equalityFieldColumns);
+ }
+ return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
+ }
+ }
+ }
+
+ private DataStream<RowData> distributeDataStreamByRangeDistributionMode(
+ DataStream<RowData> input,
+ Schema iSchema,
+ PartitionSpec partitionSpec,
+ SortOrder sortOrderParam) {
+
+ int writerParallelism =
Review Comment:
Hi @stevenzwu
Thanks for your comment.
We have the same logic in the beginning of the `distributeDataStreamByRangeDistributionMode` method:
```
int writerParallelism =
flinkWriteConf.writeParallelism() == null
? input.getParallelism()
: flinkWriteConf.writeParallelism();
```
So I think that for range partitioning the behavior should be the same. What do you think ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Range distribution iceberg sink [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#issuecomment-2887169195
+1 it would be nice to follow up with this. If @rodmeneses is busy, maybe @Guosmilesmile could also take this one?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Range distribution iceberg sink [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#discussion_r1946496791
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2DistributionMode.java:
##########
@@ -0,0 +1,495 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.sink;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assumptions.assumeThat;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.apache.flink.types.Row;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DistributionMode;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Parameter;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.flink.FlinkWriteOptions;
+import org.apache.iceberg.flink.HadoopCatalogExtension;
+import org.apache.iceberg.flink.MiniFlinkClusterExtension;
+import org.apache.iceberg.flink.SimpleDataUtil;
+import org.apache.iceberg.flink.TestFixtures;
+import org.apache.iceberg.flink.sink.shuffle.StatisticsType;
+import org.apache.iceberg.flink.source.BoundedTestSource;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.types.Conversions;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+/**
+ * This tests the distribution mode of the IcebergSink. Extract them separately since it is
+ * unnecessary to test different file formats (Avro, Orc, Parquet) like in {@link
+ * TestFlinkIcebergSink}. Removing the file format dimension reduces the number of combinations from
+ * 12 to 4, which helps reduce test run time.
+ */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestFlinkIcebergSinkV2DistributionMode extends TestFlinkIcebergSinkBase {
+
+ @RegisterExtension
+ public static MiniClusterExtension miniClusterResource =
+ MiniFlinkClusterExtension.createWithClassloaderCheckDisabled();
+
+ @RegisterExtension
+ private static final HadoopCatalogExtension CATALOG_EXTENSION =
+ new HadoopCatalogExtension(TestFixtures.DATABASE, TestFixtures.TABLE);
+
+ private final FileFormat format = FileFormat.PARQUET;
+
+ @Parameter(index = 0)
+ private int parallelism;
+
+ @Parameter(index = 1)
+ private boolean partitioned;
+
+ @Parameters(name = "parallelism = {0}, partitioned = {1}")
+ public static Object[][] parameters() {
+ return new Object[][] {
+ {1, true},
+ {1, false},
+ {2, true},
+ {2, false}
+ };
+ }
+
+ @BeforeEach
+ public void before() throws IOException {
+ this.table =
+ CATALOG_EXTENSION
+ .catalog()
+ .createTable(
+ TestFixtures.TABLE_IDENTIFIER,
+ SimpleDataUtil.SCHEMA,
+ partitioned
+ ? PartitionSpec.builderFor(SimpleDataUtil.SCHEMA).identity("data").build()
+ : PartitionSpec.unpartitioned(),
+ ImmutableMap.of(TableProperties.DEFAULT_FILE_FORMAT, format.name()));
+
+ this.env =
+ StreamExecutionEnvironment.getExecutionEnvironment(
+ MiniFlinkClusterExtension.DISABLE_CLASSLOADER_CHECK_CONFIG)
+ .enableCheckpointing(100)
+ .setParallelism(parallelism)
+ .setMaxParallelism(parallelism);
+
+ this.tableLoader = CATALOG_EXTENSION.tableLoader();
+ }
+
+ @TestTemplate
+ public void testShuffleByPartitionWithSchema() throws Exception {
+ testWriteRow(parallelism, SimpleDataUtil.FLINK_SCHEMA, DistributionMode.HASH);
+ if (partitioned) {
+ assertThat(partitionFiles("aaa")).isEqualTo(1);
+ assertThat(partitionFiles("bbb")).isEqualTo(1);
+ assertThat(partitionFiles("ccc")).isEqualTo(1);
+ }
+ }
+
+ @TestTemplate
+ public void testJobNoneDistributeMode() throws Exception {
+ table
+ .updateProperties()
+ .set(TableProperties.WRITE_DISTRIBUTION_MODE, DistributionMode.HASH.modeName())
+ .commit();
+
+ testWriteRow(parallelism, null, DistributionMode.NONE);
+
+ if (parallelism > 1) {
+ if (partitioned) {
+ int files = partitionFiles("aaa") + partitionFiles("bbb") + partitionFiles("ccc");
+ assertThat(files).isGreaterThan(3);
+ }
+ }
+ }
+
+ @TestTemplate
+ public void testJobNullDistributionMode() throws Exception {
+ table
+ .updateProperties()
+ .set(TableProperties.WRITE_DISTRIBUTION_MODE, DistributionMode.HASH.modeName())
+ .commit();
+
+ testWriteRow(parallelism, null, null);
+
+ if (partitioned) {
+ assertThat(partitionFiles("aaa")).isEqualTo(1);
+ assertThat(partitionFiles("bbb")).isEqualTo(1);
+ assertThat(partitionFiles("ccc")).isEqualTo(1);
+ }
+ }
+
+ @TestTemplate
+ public void testPartitionWriteMode() throws Exception {
+ testWriteRow(parallelism, null, DistributionMode.HASH);
+ if (partitioned) {
+ assertThat(partitionFiles("aaa")).isEqualTo(1);
+ assertThat(partitionFiles("bbb")).isEqualTo(1);
+ assertThat(partitionFiles("ccc")).isEqualTo(1);
+ }
+ }
+
+ @TestTemplate
+ public void testOverrideWriteConfigWithUnknownDistributionMode() {
+ Map<String, String> newProps = Maps.newHashMap();
+ newProps.put(FlinkWriteOptions.DISTRIBUTION_MODE.key(), "UNRECOGNIZED");
+
+ List<Row> rows = createRows("");
+ DataStream<Row> dataStream = env.addSource(createBoundedSource(rows), ROW_TYPE_INFO);
+
+ IcebergSink.forRow(dataStream, SimpleDataUtil.FLINK_SCHEMA)
+ .table(table)
+ .tableLoader(tableLoader)
+ .writeParallelism(parallelism)
+ .setAll(newProps)
+ .append();
+
+ assertThatThrownBy(env::execute)
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid distribution mode: UNRECOGNIZED");
+ }
+
+ @TestTemplate
+ public void testRangeDistributionWithoutSortOrderUnpartitioned() throws Exception {
+ assumeThat(partitioned).isFalse();
+
+ table
+ .updateProperties()
+ .set(TableProperties.WRITE_DISTRIBUTION_MODE, DistributionMode.RANGE.modeName())
+ .commit();
+
+ int numOfCheckpoints = 6;
+ DataStream<Row> dataStream =
+ env.addSource(
+ createRangeDistributionBoundedSource(createCharRows(numOfCheckpoints, 10)),
+ ROW_TYPE_INFO);
+
+ IcebergSink.forRow(dataStream, SimpleDataUtil.FLINK_SCHEMA)
+ .table(table)
+ .tableLoader(tableLoader)
+ .writeParallelism(parallelism)
+ .append();
+
+ // Range distribution requires either sort order or partition spec defined
+ assertThatThrownBy(env::execute)
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage(
+ "Invalid write distribution mode: range. Need to define sort order or partition spec.");
+ }
+
+ @TestTemplate
+ public void testRangeDistributionWithoutSortOrderPartitioned() throws Exception {
+ assumeThat(partitioned).isTrue();
+
+ table
+ .updateProperties()
+ .set(TableProperties.WRITE_DISTRIBUTION_MODE, DistributionMode.RANGE.modeName())
+ .commit();
+
+ int numOfCheckpoints = 6;
+ DataStream<Row> dataStream =
+ env.addSource(
+ createRangeDistributionBoundedSource(createCharRows(numOfCheckpoints, 10)),
+ ROW_TYPE_INFO);
+ IcebergSink.Builder builder =
+ IcebergSink.forRow(dataStream, SimpleDataUtil.FLINK_SCHEMA)
+ .table(table)
+ .tableLoader(tableLoader)
+ .writeParallelism(parallelism);
+
+ // sort based on partition columns
+ builder.append();
+ env.execute(getClass().getSimpleName());
+
+ table.refresh();
+ // ordered in reverse timeline from the newest snapshot to the oldest snapshot
+ List<Snapshot> snapshots = Lists.newArrayList(table.snapshots().iterator());
+ // only keep the snapshots with added data files
+ snapshots =
+ snapshots.stream()
+ .filter(snapshot -> snapshot.addedDataFiles(table.io()).iterator().hasNext())
+ .collect(Collectors.toList());
+
+ // Sometimes we will have more checkpoints than the bounded source if we pass the
+ // auto checkpoint interval. Thus producing multiple snapshots.
+ assertThat(snapshots).hasSizeGreaterThanOrEqualTo(numOfCheckpoints);
Review Comment:
How likely is that this test fails if the distribution mode is not range?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Range distribution iceberg sink [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#issuecomment-2709153324
This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: port range distribution to v2 iceberg sink [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#discussion_r2101587781
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -645,72 +711,135 @@ private DataStream<RowData> distributeDataStream(DataStream<RowData> input) {
DistributionMode mode = flinkWriteConf.distributionMode();
Schema schema = table.schema();
PartitionSpec spec = table.spec();
+ SortOrder sortOrder = table.sortOrder();
+
LOG.info("Write distribution mode is '{}'", mode.modeName());
switch (mode) {
case NONE:
- if (equalityFieldIds.isEmpty()) {
- return input;
- } else {
- LOG.info("Distribute rows by equality fields, because there are equality fields set");
- return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- }
-
+ return distributeDataStreamByNoneDistributionMode(input, schema);
case HASH:
- if (equalityFieldIds.isEmpty()) {
- if (table.spec().isUnpartitioned()) {
- LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and table is unpartitioned");
- return input;
- } else {
- if (BucketPartitionerUtil.hasOneBucketField(spec)) {
- return input.partitionCustom(
- new BucketPartitioner(spec),
- new BucketPartitionKeySelector(spec, schema, flinkRowType));
- } else {
- return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
- }
- }
- } else {
- if (spec.isUnpartitioned()) {
- LOG.info(
- "Distribute rows by equality fields, because there are equality fields set "
- + "and table is unpartitioned");
- return input.keyBy(
- new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- } else {
- for (PartitionField partitionField : spec.fields()) {
- Preconditions.checkState(
- equalityFieldIds.contains(partitionField.sourceId()),
- "In 'hash' distribution mode with equality fields set, partition field '%s' "
- + "should be included in equality fields: '%s'",
- partitionField,
- equalityFieldColumns);
- }
- return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
- }
- }
-
+ return distributeDataStreamByHashDistributionMode(input, schema, spec);
case RANGE:
- if (equalityFieldIds.isEmpty()) {
- LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and {}=range is not supported yet in flink",
- WRITE_DISTRIBUTION_MODE);
- return input;
- } else {
- LOG.info(
- "Distribute rows by equality fields, because there are equality fields set "
- + "and{}=range is not supported yet in flink",
- WRITE_DISTRIBUTION_MODE);
- return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- }
-
+ return distributeDataStreamByRangeDistributionMode(input, schema, spec, sortOrder);
default:
throw new RuntimeException("Unrecognized " + WRITE_DISTRIBUTION_MODE + ": " + mode);
}
}
+ private DataStream<RowData> distributeDataStreamByNoneDistributionMode(
+ DataStream<RowData> input, Schema iSchema) {
+ if (equalityFieldIds.isEmpty()) {
+ return input;
+ } else {
+ LOG.info("Distribute rows by equality fields, because there are equality fields set");
+ return input.keyBy(new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));
+ }
+ }
+
+ private DataStream<RowData> distributeDataStreamByHashDistributionMode(
+ DataStream<RowData> input, Schema iSchema, PartitionSpec partitionSpec) {
+ if (equalityFieldIds.isEmpty()) {
+ if (partitionSpec.isUnpartitioned()) {
+ LOG.warn(
+ "Fallback to use 'none' distribution mode, because there are no equality fields set "
+ + "and table is unpartitioned");
+ return input;
+ } else {
+ return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
+ }
+ } else {
+ if (partitionSpec.isUnpartitioned()) {
+ LOG.info(
+ "Distribute rows by equality fields, because there are equality fields set "
+ + "and table is unpartitioned");
+ return input.keyBy(new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));
+ } else {
+ for (PartitionField partitionField : partitionSpec.fields()) {
+ Preconditions.checkState(
+ equalityFieldIds.contains(partitionField.sourceId()),
+ "In 'hash' distribution mode with equality fields set, partition field '%s' "
+ + "should be included in equality fields: '%s'",
+ partitionField,
+ equalityFieldColumns);
+ }
+ return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
+ }
+ }
+ }
+
+ private DataStream<RowData> distributeDataStreamByRangeDistributionMode(
+ DataStream<RowData> input,
+ Schema iSchema,
+ PartitionSpec partitionSpec,
+ SortOrder sortOrderParam) {
+
+ int writerParallelism =
Review Comment:
this logic should be applied to the writer parallelism for v2 sink.
if write parallelism is not configured, the v1 sink default the writer parallelism to the input parallelism to promote chaining. Want to confirm if that is the case for v2 sink? from reading the code, I thought v2 sink will default the writer parallelism to default job parallelism?
```
// Note that IcebergSink internally consists o multiple operators (like writer, committer,
// aggregator).
// The following parallelism will be propagated to all of the above operators.
if (sink.flinkWriteConf.writeParallelism() != null) {
rowDataDataStreamSink.setParallelism(sink.flinkWriteConf.writeParallelism());
}
```
technically, if this is a behavior change problem for v2 sink, it is not caused by this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Range distribution iceberg sink [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#issuecomment-2799347878
This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Range distribution iceberg sink [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#discussion_r1946498357
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSink.java:
##########
@@ -140,6 +140,13 @@ void testWriteRowWithTableSchema() throws Exception {
testWriteRow(SimpleDataUtil.FLINK_SCHEMA, DistributionMode.NONE);
}
+ @TestTemplate
+ void testWriteRowWithTableSchemaRANGE() throws Exception {
+ if (partitioned) {
+ testWriteRow(SimpleDataUtil.FLINK_SCHEMA, DistributionMode.RANGE);
Review Comment:
Can we check the job graph to verify that the partitioner is set?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: port range distribution to v2 iceberg sink [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#discussion_r2101587781
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -645,72 +711,135 @@ private DataStream<RowData> distributeDataStream(DataStream<RowData> input) {
DistributionMode mode = flinkWriteConf.distributionMode();
Schema schema = table.schema();
PartitionSpec spec = table.spec();
+ SortOrder sortOrder = table.sortOrder();
+
LOG.info("Write distribution mode is '{}'", mode.modeName());
switch (mode) {
case NONE:
- if (equalityFieldIds.isEmpty()) {
- return input;
- } else {
- LOG.info("Distribute rows by equality fields, because there are equality fields set");
- return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- }
-
+ return distributeDataStreamByNoneDistributionMode(input, schema);
case HASH:
- if (equalityFieldIds.isEmpty()) {
- if (table.spec().isUnpartitioned()) {
- LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and table is unpartitioned");
- return input;
- } else {
- if (BucketPartitionerUtil.hasOneBucketField(spec)) {
- return input.partitionCustom(
- new BucketPartitioner(spec),
- new BucketPartitionKeySelector(spec, schema, flinkRowType));
- } else {
- return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
- }
- }
- } else {
- if (spec.isUnpartitioned()) {
- LOG.info(
- "Distribute rows by equality fields, because there are equality fields set "
- + "and table is unpartitioned");
- return input.keyBy(
- new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- } else {
- for (PartitionField partitionField : spec.fields()) {
- Preconditions.checkState(
- equalityFieldIds.contains(partitionField.sourceId()),
- "In 'hash' distribution mode with equality fields set, partition field '%s' "
- + "should be included in equality fields: '%s'",
- partitionField,
- equalityFieldColumns);
- }
- return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
- }
- }
-
+ return distributeDataStreamByHashDistributionMode(input, schema, spec);
case RANGE:
- if (equalityFieldIds.isEmpty()) {
- LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and {}=range is not supported yet in flink",
- WRITE_DISTRIBUTION_MODE);
- return input;
- } else {
- LOG.info(
- "Distribute rows by equality fields, because there are equality fields set "
- + "and{}=range is not supported yet in flink",
- WRITE_DISTRIBUTION_MODE);
- return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- }
-
+ return distributeDataStreamByRangeDistributionMode(input, schema, spec, sortOrder);
default:
throw new RuntimeException("Unrecognized " + WRITE_DISTRIBUTION_MODE + ": " + mode);
}
}
+ private DataStream<RowData> distributeDataStreamByNoneDistributionMode(
+ DataStream<RowData> input, Schema iSchema) {
+ if (equalityFieldIds.isEmpty()) {
+ return input;
+ } else {
+ LOG.info("Distribute rows by equality fields, because there are equality fields set");
+ return input.keyBy(new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));
+ }
+ }
+
+ private DataStream<RowData> distributeDataStreamByHashDistributionMode(
+ DataStream<RowData> input, Schema iSchema, PartitionSpec partitionSpec) {
+ if (equalityFieldIds.isEmpty()) {
+ if (partitionSpec.isUnpartitioned()) {
+ LOG.warn(
+ "Fallback to use 'none' distribution mode, because there are no equality fields set "
+ + "and table is unpartitioned");
+ return input;
+ } else {
+ return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
+ }
+ } else {
+ if (partitionSpec.isUnpartitioned()) {
+ LOG.info(
+ "Distribute rows by equality fields, because there are equality fields set "
+ + "and table is unpartitioned");
+ return input.keyBy(new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));
+ } else {
+ for (PartitionField partitionField : partitionSpec.fields()) {
+ Preconditions.checkState(
+ equalityFieldIds.contains(partitionField.sourceId()),
+ "In 'hash' distribution mode with equality fields set, partition field '%s' "
+ + "should be included in equality fields: '%s'",
+ partitionField,
+ equalityFieldColumns);
+ }
+ return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
+ }
+ }
+ }
+
+ private DataStream<RowData> distributeDataStreamByRangeDistributionMode(
+ DataStream<RowData> input,
+ Schema iSchema,
+ PartitionSpec partitionSpec,
+ SortOrder sortOrderParam) {
+
+ int writerParallelism =
Review Comment:
this logic should be applied to the writer parallelism for v2 sink.
if write parallelism is not configured, the v1 sink default the writer parallelism to the input parallelism to promote chaining. Want to confirm if that is the case for v2 sink? from reading the code, I thought v2 sink will default the writer parallelism to default job parallelism?
```
// Note that IcebergSink internally consists o multiple operators (like writer, committer,
// aggregator).
// The following parallelism will be propagated to all of the above operators.
if (sink.flinkWriteConf.writeParallelism() != null) {
rowDataDataStreamSink.setParallelism(sink.flinkWriteConf.writeParallelism());
}
```
technically, if this is a behavior change problem for v2 sink, it is not caused by this PR. but it is critical that the same writer parallelism is used by the shuffle operator to properly range partition the data to downstream writer tasks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: port range distribution to v2 iceberg sink [iceberg]
Posted by "rodmeneses (via GitHub)" <gi...@apache.org>.
rodmeneses commented on code in PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#discussion_r2112131384
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -645,72 +711,135 @@ private DataStream<RowData> distributeDataStream(DataStream<RowData> input) {
DistributionMode mode = flinkWriteConf.distributionMode();
Schema schema = table.schema();
PartitionSpec spec = table.spec();
+ SortOrder sortOrder = table.sortOrder();
+
LOG.info("Write distribution mode is '{}'", mode.modeName());
switch (mode) {
case NONE:
- if (equalityFieldIds.isEmpty()) {
- return input;
- } else {
- LOG.info("Distribute rows by equality fields, because there are equality fields set");
- return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- }
-
+ return distributeDataStreamByNoneDistributionMode(input, schema);
case HASH:
- if (equalityFieldIds.isEmpty()) {
- if (table.spec().isUnpartitioned()) {
- LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and table is unpartitioned");
- return input;
- } else {
- if (BucketPartitionerUtil.hasOneBucketField(spec)) {
- return input.partitionCustom(
- new BucketPartitioner(spec),
- new BucketPartitionKeySelector(spec, schema, flinkRowType));
- } else {
- return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
- }
- }
- } else {
- if (spec.isUnpartitioned()) {
- LOG.info(
- "Distribute rows by equality fields, because there are equality fields set "
- + "and table is unpartitioned");
- return input.keyBy(
- new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- } else {
- for (PartitionField partitionField : spec.fields()) {
- Preconditions.checkState(
- equalityFieldIds.contains(partitionField.sourceId()),
- "In 'hash' distribution mode with equality fields set, partition field '%s' "
- + "should be included in equality fields: '%s'",
- partitionField,
- equalityFieldColumns);
- }
- return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
- }
- }
-
+ return distributeDataStreamByHashDistributionMode(input, schema, spec);
case RANGE:
- if (equalityFieldIds.isEmpty()) {
- LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and {}=range is not supported yet in flink",
- WRITE_DISTRIBUTION_MODE);
- return input;
- } else {
- LOG.info(
- "Distribute rows by equality fields, because there are equality fields set "
- + "and{}=range is not supported yet in flink",
- WRITE_DISTRIBUTION_MODE);
- return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- }
-
+ return distributeDataStreamByRangeDistributionMode(input, schema, spec, sortOrder);
default:
throw new RuntimeException("Unrecognized " + WRITE_DISTRIBUTION_MODE + ": " + mode);
}
}
+ private DataStream<RowData> distributeDataStreamByNoneDistributionMode(
+ DataStream<RowData> input, Schema iSchema) {
+ if (equalityFieldIds.isEmpty()) {
+ return input;
+ } else {
+ LOG.info("Distribute rows by equality fields, because there are equality fields set");
+ return input.keyBy(new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));
+ }
+ }
+
+ private DataStream<RowData> distributeDataStreamByHashDistributionMode(
+ DataStream<RowData> input, Schema iSchema, PartitionSpec partitionSpec) {
+ if (equalityFieldIds.isEmpty()) {
+ if (partitionSpec.isUnpartitioned()) {
+ LOG.warn(
+ "Fallback to use 'none' distribution mode, because there are no equality fields set "
+ + "and table is unpartitioned");
+ return input;
+ } else {
+ return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
+ }
+ } else {
+ if (partitionSpec.isUnpartitioned()) {
+ LOG.info(
+ "Distribute rows by equality fields, because there are equality fields set "
+ + "and table is unpartitioned");
+ return input.keyBy(new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));
+ } else {
+ for (PartitionField partitionField : partitionSpec.fields()) {
+ Preconditions.checkState(
+ equalityFieldIds.contains(partitionField.sourceId()),
+ "In 'hash' distribution mode with equality fields set, partition field '%s' "
+ + "should be included in equality fields: '%s'",
+ partitionField,
+ equalityFieldColumns);
+ }
+ return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
+ }
+ }
+ }
+
+ private DataStream<RowData> distributeDataStreamByRangeDistributionMode(
+ DataStream<RowData> input,
+ Schema iSchema,
+ PartitionSpec partitionSpec,
+ SortOrder sortOrderParam) {
+
+ int writerParallelism =
Review Comment:
Hi @stevenzwu . Thanks for the clarification. I think I know what you mean. On FlinkSink, even without talking about RANGE distribution, the parallelism of the v1 sink by default will be the same as the input source parallelism.
This is a good approach, because it encourage chaining.
However, we dont have that logic on the IcebergSink v2. I think we should have the same logic there. I could do that in another PR and the follow up with this one. What do you think? Thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Range distribution iceberg sink [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#issuecomment-2612010445
Hey @rodmeneses! Thanks for porting this feature over. I'll have a look shortly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Range distribution iceberg sink [iceberg]
Posted by "rodmeneses (via GitHub)" <gi...@apache.org>.
rodmeneses commented on PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#issuecomment-2722917905
keeping it alive
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Range distribution iceberg sink [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#discussion_r1946494527
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2DistributionMode.java:
##########
@@ -0,0 +1,495 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.sink;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assumptions.assumeThat;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.apache.flink.types.Row;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DistributionMode;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Parameter;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.flink.FlinkWriteOptions;
+import org.apache.iceberg.flink.HadoopCatalogExtension;
+import org.apache.iceberg.flink.MiniFlinkClusterExtension;
+import org.apache.iceberg.flink.SimpleDataUtil;
+import org.apache.iceberg.flink.TestFixtures;
+import org.apache.iceberg.flink.sink.shuffle.StatisticsType;
+import org.apache.iceberg.flink.source.BoundedTestSource;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.types.Conversions;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+/**
+ * This tests the distribution mode of the IcebergSink. Extract them separately since it is
+ * unnecessary to test different file formats (Avro, Orc, Parquet) like in {@link
+ * TestFlinkIcebergSink}. Removing the file format dimension reduces the number of combinations from
+ * 12 to 4, which helps reduce test run time.
+ */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestFlinkIcebergSinkV2DistributionMode extends TestFlinkIcebergSinkBase {
+
+ @RegisterExtension
+ public static MiniClusterExtension miniClusterResource =
+ MiniFlinkClusterExtension.createWithClassloaderCheckDisabled();
+
+ @RegisterExtension
+ private static final HadoopCatalogExtension CATALOG_EXTENSION =
+ new HadoopCatalogExtension(TestFixtures.DATABASE, TestFixtures.TABLE);
+
+ private final FileFormat format = FileFormat.PARQUET;
+
+ @Parameter(index = 0)
+ private int parallelism;
+
+ @Parameter(index = 1)
+ private boolean partitioned;
+
+ @Parameters(name = "parallelism = {0}, partitioned = {1}")
+ public static Object[][] parameters() {
+ return new Object[][] {
+ {1, true},
+ {1, false},
+ {2, true},
+ {2, false}
+ };
+ }
+
+ @BeforeEach
+ public void before() throws IOException {
+ this.table =
+ CATALOG_EXTENSION
+ .catalog()
+ .createTable(
+ TestFixtures.TABLE_IDENTIFIER,
+ SimpleDataUtil.SCHEMA,
+ partitioned
+ ? PartitionSpec.builderFor(SimpleDataUtil.SCHEMA).identity("data").build()
+ : PartitionSpec.unpartitioned(),
+ ImmutableMap.of(TableProperties.DEFAULT_FILE_FORMAT, format.name()));
+
+ this.env =
+ StreamExecutionEnvironment.getExecutionEnvironment(
+ MiniFlinkClusterExtension.DISABLE_CLASSLOADER_CHECK_CONFIG)
+ .enableCheckpointing(100)
+ .setParallelism(parallelism)
+ .setMaxParallelism(parallelism);
+
+ this.tableLoader = CATALOG_EXTENSION.tableLoader();
+ }
+
+ @TestTemplate
+ public void testShuffleByPartitionWithSchema() throws Exception {
+ testWriteRow(parallelism, SimpleDataUtil.FLINK_SCHEMA, DistributionMode.HASH);
+ if (partitioned) {
+ assertThat(partitionFiles("aaa")).isEqualTo(1);
+ assertThat(partitionFiles("bbb")).isEqualTo(1);
+ assertThat(partitionFiles("ccc")).isEqualTo(1);
+ }
+ }
+
+ @TestTemplate
+ public void testJobNoneDistributeMode() throws Exception {
+ table
+ .updateProperties()
+ .set(TableProperties.WRITE_DISTRIBUTION_MODE, DistributionMode.HASH.modeName())
+ .commit();
+
+ testWriteRow(parallelism, null, DistributionMode.NONE);
+
+ if (parallelism > 1) {
+ if (partitioned) {
+ int files = partitionFiles("aaa") + partitionFiles("bbb") + partitionFiles("ccc");
+ assertThat(files).isGreaterThan(3);
+ }
+ }
+ }
+
+ @TestTemplate
+ public void testJobNullDistributionMode() throws Exception {
+ table
+ .updateProperties()
+ .set(TableProperties.WRITE_DISTRIBUTION_MODE, DistributionMode.HASH.modeName())
+ .commit();
+
+ testWriteRow(parallelism, null, null);
+
+ if (partitioned) {
+ assertThat(partitionFiles("aaa")).isEqualTo(1);
+ assertThat(partitionFiles("bbb")).isEqualTo(1);
+ assertThat(partitionFiles("ccc")).isEqualTo(1);
+ }
+ }
+
+ @TestTemplate
+ public void testPartitionWriteMode() throws Exception {
+ testWriteRow(parallelism, null, DistributionMode.HASH);
+ if (partitioned) {
+ assertThat(partitionFiles("aaa")).isEqualTo(1);
+ assertThat(partitionFiles("bbb")).isEqualTo(1);
+ assertThat(partitionFiles("ccc")).isEqualTo(1);
+ }
+ }
+
+ @TestTemplate
+ public void testOverrideWriteConfigWithUnknownDistributionMode() {
+ Map<String, String> newProps = Maps.newHashMap();
+ newProps.put(FlinkWriteOptions.DISTRIBUTION_MODE.key(), "UNRECOGNIZED");
+
+ List<Row> rows = createRows("");
+ DataStream<Row> dataStream = env.addSource(createBoundedSource(rows), ROW_TYPE_INFO);
+
+ IcebergSink.forRow(dataStream, SimpleDataUtil.FLINK_SCHEMA)
+ .table(table)
+ .tableLoader(tableLoader)
+ .writeParallelism(parallelism)
+ .setAll(newProps)
+ .append();
+
+ assertThatThrownBy(env::execute)
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid distribution mode: UNRECOGNIZED");
+ }
+
+ @TestTemplate
+ public void testRangeDistributionWithoutSortOrderUnpartitioned() throws Exception {
+ assumeThat(partitioned).isFalse();
+
+ table
+ .updateProperties()
+ .set(TableProperties.WRITE_DISTRIBUTION_MODE, DistributionMode.RANGE.modeName())
+ .commit();
+
+ int numOfCheckpoints = 6;
+ DataStream<Row> dataStream =
+ env.addSource(
+ createRangeDistributionBoundedSource(createCharRows(numOfCheckpoints, 10)),
+ ROW_TYPE_INFO);
+
+ IcebergSink.forRow(dataStream, SimpleDataUtil.FLINK_SCHEMA)
+ .table(table)
+ .tableLoader(tableLoader)
+ .writeParallelism(parallelism)
+ .append();
+
+ // Range distribution requires either sort order or partition spec defined
+ assertThatThrownBy(env::execute)
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage(
+ "Invalid write distribution mode: range. Need to define sort order or partition spec.");
+ }
+
+ @TestTemplate
+ public void testRangeDistributionWithoutSortOrderPartitioned() throws Exception {
+ assumeThat(partitioned).isTrue();
+
+ table
+ .updateProperties()
+ .set(TableProperties.WRITE_DISTRIBUTION_MODE, DistributionMode.RANGE.modeName())
+ .commit();
+
+ int numOfCheckpoints = 6;
+ DataStream<Row> dataStream =
+ env.addSource(
+ createRangeDistributionBoundedSource(createCharRows(numOfCheckpoints, 10)),
+ ROW_TYPE_INFO);
+ IcebergSink.Builder builder =
+ IcebergSink.forRow(dataStream, SimpleDataUtil.FLINK_SCHEMA)
+ .table(table)
+ .tableLoader(tableLoader)
+ .writeParallelism(parallelism);
+
+ // sort based on partition columns
+ builder.append();
Review Comment:
Why not chain the append?
I think the comment has nothing to do with the line
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: port range distribution to v2 iceberg sink [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu merged PR #12071:
URL: https://github.com/apache/iceberg/pull/12071
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: port range distribution to v2 iceberg sink [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#discussion_r2112952874
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -645,72 +711,135 @@ private DataStream<RowData> distributeDataStream(DataStream<RowData> input) {
DistributionMode mode = flinkWriteConf.distributionMode();
Schema schema = table.schema();
PartitionSpec spec = table.spec();
+ SortOrder sortOrder = table.sortOrder();
+
LOG.info("Write distribution mode is '{}'", mode.modeName());
switch (mode) {
case NONE:
- if (equalityFieldIds.isEmpty()) {
- return input;
- } else {
- LOG.info("Distribute rows by equality fields, because there are equality fields set");
- return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- }
-
+ return distributeDataStreamByNoneDistributionMode(input, schema);
case HASH:
- if (equalityFieldIds.isEmpty()) {
- if (table.spec().isUnpartitioned()) {
- LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and table is unpartitioned");
- return input;
- } else {
- if (BucketPartitionerUtil.hasOneBucketField(spec)) {
- return input.partitionCustom(
- new BucketPartitioner(spec),
- new BucketPartitionKeySelector(spec, schema, flinkRowType));
- } else {
- return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
- }
- }
- } else {
- if (spec.isUnpartitioned()) {
- LOG.info(
- "Distribute rows by equality fields, because there are equality fields set "
- + "and table is unpartitioned");
- return input.keyBy(
- new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- } else {
- for (PartitionField partitionField : spec.fields()) {
- Preconditions.checkState(
- equalityFieldIds.contains(partitionField.sourceId()),
- "In 'hash' distribution mode with equality fields set, partition field '%s' "
- + "should be included in equality fields: '%s'",
- partitionField,
- equalityFieldColumns);
- }
- return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
- }
- }
-
+ return distributeDataStreamByHashDistributionMode(input, schema, spec);
case RANGE:
- if (equalityFieldIds.isEmpty()) {
- LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and {}=range is not supported yet in flink",
- WRITE_DISTRIBUTION_MODE);
- return input;
- } else {
- LOG.info(
- "Distribute rows by equality fields, because there are equality fields set "
- + "and{}=range is not supported yet in flink",
- WRITE_DISTRIBUTION_MODE);
- return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- }
-
+ return distributeDataStreamByRangeDistributionMode(input, schema, spec, sortOrder);
default:
throw new RuntimeException("Unrecognized " + WRITE_DISTRIBUTION_MODE + ": " + mode);
}
}
+ private DataStream<RowData> distributeDataStreamByNoneDistributionMode(
+ DataStream<RowData> input, Schema iSchema) {
+ if (equalityFieldIds.isEmpty()) {
+ return input;
+ } else {
+ LOG.info("Distribute rows by equality fields, because there are equality fields set");
+ return input.keyBy(new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));
+ }
+ }
+
+ private DataStream<RowData> distributeDataStreamByHashDistributionMode(
+ DataStream<RowData> input, Schema iSchema, PartitionSpec partitionSpec) {
+ if (equalityFieldIds.isEmpty()) {
+ if (partitionSpec.isUnpartitioned()) {
+ LOG.warn(
+ "Fallback to use 'none' distribution mode, because there are no equality fields set "
+ + "and table is unpartitioned");
+ return input;
+ } else {
+ return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
+ }
+ } else {
+ if (partitionSpec.isUnpartitioned()) {
+ LOG.info(
+ "Distribute rows by equality fields, because there are equality fields set "
+ + "and table is unpartitioned");
+ return input.keyBy(new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));
+ } else {
+ for (PartitionField partitionField : partitionSpec.fields()) {
+ Preconditions.checkState(
+ equalityFieldIds.contains(partitionField.sourceId()),
+ "In 'hash' distribution mode with equality fields set, partition field '%s' "
+ + "should be included in equality fields: '%s'",
+ partitionField,
+ equalityFieldColumns);
+ }
+ return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
+ }
+ }
+ }
+
+ private DataStream<RowData> distributeDataStreamByRangeDistributionMode(
+ DataStream<RowData> input,
+ Schema iSchema,
+ PartitionSpec partitionSpec,
+ SortOrder sortOrderParam) {
+
+ int writerParallelism =
Review Comment:
sure. you can follow up with a new PR for the writer parallelism fix.
can you resolve the conflict? then we can merge this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Range distribution iceberg sink [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #12071: Range distribution iceberg sink
URL: https://github.com/apache/iceberg/pull/12071
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Range distribution iceberg sink [iceberg]
Posted by "rodmeneses (via GitHub)" <gi...@apache.org>.
rodmeneses commented on code in PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#discussion_r2098985225
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -692,20 +729,51 @@ private DataStream<RowData> distributeDataStream(DataStream<RowData> input) {
}
case RANGE:
- if (equalityFieldIds.isEmpty()) {
+ // Ideally, exception should be thrown in the combination of range distribution and
+ // equality fields. Primary key case should use hash distribution mode.
+ // Keep the current behavior of falling back to keyBy for backward compatibility.
+ if (!equalityFieldIds.isEmpty()) {
LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and {}=range is not supported yet in flink",
- WRITE_DISTRIBUTION_MODE);
- return input;
- } else {
- LOG.info(
- "Distribute rows by equality fields, because there are equality fields set "
- + "and{}=range is not supported yet in flink",
+ "Hash distribute rows by equality fields, even though {}=range is set. "
+ + "Range distribution for primary keys are not always safe in "
+ + "Flink streaming writer.",
WRITE_DISTRIBUTION_MODE);
return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
}
+ // range distribute by partition key or sort key if table has an SortOrder
+ Preconditions.checkState(
+ sortOrder.isSorted() || spec.isPartitioned(),
+ "Invalid write distribution mode: range. Need to define sort order or partition spec.");
+ if (sortOrder.isUnsorted()) {
+ sortOrder = Partitioning.sortOrderFor(spec);
+ LOG.info("Construct sort order from partition spec");
+ }
+
+ LOG.info("Range distribute rows by sort order: {}", sortOrder);
+ StatisticsType statisticsType = flinkWriteConf.rangeDistributionStatisticsType();
+ SingleOutputStreamOperator<StatisticsOrRecord> shuffleStream =
+ input
+ .transform(
+ operatorName("range-shuffle"),
+ TypeInformation.of(StatisticsOrRecord.class),
+ new DataStatisticsOperatorFactory(
+ schema,
+ sortOrder,
+ writerParallelism,
+ statisticsType,
+ flinkWriteConf.rangeDistributionSortKeyBaseWeight()))
+ // Set the parallelism same as input operator to encourage chaining
+ .setParallelism(input.getParallelism());
Review Comment:
I added slotSharingGroup to enforce
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Range distribution iceberg sink [iceberg]
Posted by "rodmeneses (via GitHub)" <gi...@apache.org>.
rodmeneses commented on code in PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#discussion_r2098898675
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2DistributionMode.java:
##########
@@ -0,0 +1,495 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.sink;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assumptions.assumeThat;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.apache.flink.types.Row;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DistributionMode;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Parameter;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.flink.FlinkWriteOptions;
+import org.apache.iceberg.flink.HadoopCatalogExtension;
+import org.apache.iceberg.flink.MiniFlinkClusterExtension;
+import org.apache.iceberg.flink.SimpleDataUtil;
+import org.apache.iceberg.flink.TestFixtures;
+import org.apache.iceberg.flink.sink.shuffle.StatisticsType;
+import org.apache.iceberg.flink.source.BoundedTestSource;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.types.Conversions;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+/**
+ * This tests the distribution mode of the IcebergSink. Extract them separately since it is
+ * unnecessary to test different file formats (Avro, Orc, Parquet) like in {@link
+ * TestFlinkIcebergSink}. Removing the file format dimension reduces the number of combinations from
+ * 12 to 4, which helps reduce test run time.
+ */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestFlinkIcebergSinkV2DistributionMode extends TestFlinkIcebergSinkBase {
+
+ @RegisterExtension
+ public static MiniClusterExtension miniClusterResource =
+ MiniFlinkClusterExtension.createWithClassloaderCheckDisabled();
+
+ @RegisterExtension
+ private static final HadoopCatalogExtension CATALOG_EXTENSION =
+ new HadoopCatalogExtension(TestFixtures.DATABASE, TestFixtures.TABLE);
+
+ private final FileFormat format = FileFormat.PARQUET;
+
+ @Parameter(index = 0)
+ private int parallelism;
+
+ @Parameter(index = 1)
+ private boolean partitioned;
+
+ @Parameters(name = "parallelism = {0}, partitioned = {1}")
+ public static Object[][] parameters() {
+ return new Object[][] {
+ {1, true},
+ {1, false},
+ {2, true},
+ {2, false}
+ };
+ }
+
+ @BeforeEach
+ public void before() throws IOException {
+ this.table =
+ CATALOG_EXTENSION
+ .catalog()
+ .createTable(
+ TestFixtures.TABLE_IDENTIFIER,
+ SimpleDataUtil.SCHEMA,
+ partitioned
+ ? PartitionSpec.builderFor(SimpleDataUtil.SCHEMA).identity("data").build()
+ : PartitionSpec.unpartitioned(),
+ ImmutableMap.of(TableProperties.DEFAULT_FILE_FORMAT, format.name()));
+
+ this.env =
+ StreamExecutionEnvironment.getExecutionEnvironment(
+ MiniFlinkClusterExtension.DISABLE_CLASSLOADER_CHECK_CONFIG)
+ .enableCheckpointing(100)
+ .setParallelism(parallelism)
+ .setMaxParallelism(parallelism);
+
+ this.tableLoader = CATALOG_EXTENSION.tableLoader();
+ }
+
+ @TestTemplate
+ public void testShuffleByPartitionWithSchema() throws Exception {
+ testWriteRow(parallelism, SimpleDataUtil.FLINK_SCHEMA, DistributionMode.HASH);
+ if (partitioned) {
+ assertThat(partitionFiles("aaa")).isEqualTo(1);
+ assertThat(partitionFiles("bbb")).isEqualTo(1);
+ assertThat(partitionFiles("ccc")).isEqualTo(1);
+ }
+ }
+
+ @TestTemplate
+ public void testJobNoneDistributeMode() throws Exception {
+ table
+ .updateProperties()
+ .set(TableProperties.WRITE_DISTRIBUTION_MODE, DistributionMode.HASH.modeName())
+ .commit();
+
+ testWriteRow(parallelism, null, DistributionMode.NONE);
+
+ if (parallelism > 1) {
+ if (partitioned) {
+ int files = partitionFiles("aaa") + partitionFiles("bbb") + partitionFiles("ccc");
+ assertThat(files).isGreaterThan(3);
+ }
+ }
+ }
+
+ @TestTemplate
+ public void testJobNullDistributionMode() throws Exception {
+ table
+ .updateProperties()
+ .set(TableProperties.WRITE_DISTRIBUTION_MODE, DistributionMode.HASH.modeName())
+ .commit();
+
+ testWriteRow(parallelism, null, null);
+
+ if (partitioned) {
+ assertThat(partitionFiles("aaa")).isEqualTo(1);
+ assertThat(partitionFiles("bbb")).isEqualTo(1);
+ assertThat(partitionFiles("ccc")).isEqualTo(1);
+ }
+ }
+
+ @TestTemplate
+ public void testPartitionWriteMode() throws Exception {
+ testWriteRow(parallelism, null, DistributionMode.HASH);
+ if (partitioned) {
+ assertThat(partitionFiles("aaa")).isEqualTo(1);
+ assertThat(partitionFiles("bbb")).isEqualTo(1);
+ assertThat(partitionFiles("ccc")).isEqualTo(1);
+ }
+ }
+
+ @TestTemplate
+ public void testOverrideWriteConfigWithUnknownDistributionMode() {
+ Map<String, String> newProps = Maps.newHashMap();
+ newProps.put(FlinkWriteOptions.DISTRIBUTION_MODE.key(), "UNRECOGNIZED");
+
+ List<Row> rows = createRows("");
+ DataStream<Row> dataStream = env.addSource(createBoundedSource(rows), ROW_TYPE_INFO);
+
+ IcebergSink.forRow(dataStream, SimpleDataUtil.FLINK_SCHEMA)
+ .table(table)
+ .tableLoader(tableLoader)
+ .writeParallelism(parallelism)
+ .setAll(newProps)
+ .append();
+
+ assertThatThrownBy(env::execute)
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid distribution mode: UNRECOGNIZED");
+ }
+
+ @TestTemplate
+ public void testRangeDistributionWithoutSortOrderUnpartitioned() throws Exception {
+ assumeThat(partitioned).isFalse();
+
+ table
+ .updateProperties()
+ .set(TableProperties.WRITE_DISTRIBUTION_MODE, DistributionMode.RANGE.modeName())
+ .commit();
+
+ int numOfCheckpoints = 6;
+ DataStream<Row> dataStream =
+ env.addSource(
+ createRangeDistributionBoundedSource(createCharRows(numOfCheckpoints, 10)),
+ ROW_TYPE_INFO);
+
+ IcebergSink.forRow(dataStream, SimpleDataUtil.FLINK_SCHEMA)
+ .table(table)
+ .tableLoader(tableLoader)
+ .writeParallelism(parallelism)
+ .append();
+
+ // Range distribution requires either sort order or partition spec defined
+ assertThatThrownBy(env::execute)
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage(
+ "Invalid write distribution mode: range. Need to define sort order or partition spec.");
+ }
+
+ @TestTemplate
+ public void testRangeDistributionWithoutSortOrderPartitioned() throws Exception {
+ assumeThat(partitioned).isTrue();
+
+ table
+ .updateProperties()
+ .set(TableProperties.WRITE_DISTRIBUTION_MODE, DistributionMode.RANGE.modeName())
+ .commit();
+
+ int numOfCheckpoints = 6;
+ DataStream<Row> dataStream =
+ env.addSource(
+ createRangeDistributionBoundedSource(createCharRows(numOfCheckpoints, 10)),
+ ROW_TYPE_INFO);
+ IcebergSink.Builder builder =
+ IcebergSink.forRow(dataStream, SimpleDataUtil.FLINK_SCHEMA)
+ .table(table)
+ .tableLoader(tableLoader)
+ .writeParallelism(parallelism);
+
+ // sort based on partition columns
+ builder.append();
+ env.execute(getClass().getSimpleName());
+
+ table.refresh();
+ // ordered in reverse timeline from the newest snapshot to the oldest snapshot
+ List<Snapshot> snapshots = Lists.newArrayList(table.snapshots().iterator());
+ // only keep the snapshots with added data files
+ snapshots =
+ snapshots.stream()
+ .filter(snapshot -> snapshot.addedDataFiles(table.io()).iterator().hasNext())
+ .collect(Collectors.toList());
+
+ // Sometimes we will have more checkpoints than the bounded source if we pass the
+ // auto checkpoint interval. Thus producing multiple snapshots.
+ assertThat(snapshots).hasSizeGreaterThanOrEqualTo(numOfCheckpoints);
Review Comment:
I tried the same test with NONE and HASH and they were all passing as well.
I think that under this test conditions (no sort order and partitioned) all the 3 distribution modes ended up creating the same number of snapshots.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Range distribution iceberg sink [iceberg]
Posted by "rodmeneses (via GitHub)" <gi...@apache.org>.
rodmeneses commented on code in PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#discussion_r2098984273
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSink.java:
##########
@@ -140,6 +140,13 @@ void testWriteRowWithTableSchema() throws Exception {
testWriteRow(SimpleDataUtil.FLINK_SCHEMA, DistributionMode.NONE);
}
+ @TestTemplate
+ void testWriteRowWithTableSchemaRANGE() throws Exception {
+ if (partitioned) {
+ testWriteRow(SimpleDataUtil.FLINK_SCHEMA, DistributionMode.RANGE);
Review Comment:
not in this same unit test, as `env.execute` traverse the graph.
I did implement it in another unit test. But it is hacky as there's no type safe way to get an instance of the underlying `Partitioner` that is wrapped around a `CustomPartitionerWrapper`. However, with reflection magic, it is possible. I did see precedent of other unit tests leveraging Java Reflection. So, I think this is helpful, but please let me know your thoughts @pvary
```java
@TestTemplate
void testRangePartitionedIsSet() throws Exception {
if (partitioned) {
List<Row> rows = createRows("");
DataStream<Row> dataStream =
env.addSource(createBoundedSource(rows), ROW_TYPE_INFO).uid("mySourceId");
IcebergSink.forRow(dataStream, SimpleDataUtil.FLINK_SCHEMA)
.table(table)
.tableLoader(tableLoader)
.tableSchema(SimpleDataUtil.FLINK_SCHEMA)
.writeParallelism(parallelism)
.distributionMode(DistributionMode.RANGE)
.append();
boolean found = false;
for (StreamNode node : env.getStreamGraph().getStreamNodes()) {
for (StreamEdge edge : node.getOutEdges()) {
if (edge.getPartitioner() instanceof CustomPartitionerWrapper) {
CustomPartitionerWrapper wrapper = (CustomPartitionerWrapper) edge.getPartitioner();
try {
Field privatePartitionerField = CustomPartitionerWrapper.class.getDeclaredField("partitioner");
privatePartitionerField.setAccessible(true);
Object wrappedPartitionerObject = privatePartitionerField.get(wrapper);
if (wrappedPartitionerObject instanceof RangePartitioner) {
found = true;
}
} catch (Exception e) {
throw new RuntimeException(e);
}
}
}
}
assertThat(found)
.as("RangePartitioner was found")
.isTrue();
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: port range distribution to v2 iceberg sink [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#discussion_r2103486896
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -645,72 +711,135 @@ private DataStream<RowData> distributeDataStream(DataStream<RowData> input) {
DistributionMode mode = flinkWriteConf.distributionMode();
Schema schema = table.schema();
PartitionSpec spec = table.spec();
+ SortOrder sortOrder = table.sortOrder();
+
LOG.info("Write distribution mode is '{}'", mode.modeName());
switch (mode) {
case NONE:
- if (equalityFieldIds.isEmpty()) {
- return input;
- } else {
- LOG.info("Distribute rows by equality fields, because there are equality fields set");
- return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- }
-
+ return distributeDataStreamByNoneDistributionMode(input, schema);
case HASH:
- if (equalityFieldIds.isEmpty()) {
- if (table.spec().isUnpartitioned()) {
- LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and table is unpartitioned");
- return input;
- } else {
- if (BucketPartitionerUtil.hasOneBucketField(spec)) {
- return input.partitionCustom(
- new BucketPartitioner(spec),
- new BucketPartitionKeySelector(spec, schema, flinkRowType));
- } else {
- return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
- }
- }
- } else {
- if (spec.isUnpartitioned()) {
- LOG.info(
- "Distribute rows by equality fields, because there are equality fields set "
- + "and table is unpartitioned");
- return input.keyBy(
- new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- } else {
- for (PartitionField partitionField : spec.fields()) {
- Preconditions.checkState(
- equalityFieldIds.contains(partitionField.sourceId()),
- "In 'hash' distribution mode with equality fields set, partition field '%s' "
- + "should be included in equality fields: '%s'",
- partitionField,
- equalityFieldColumns);
- }
- return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
- }
- }
-
+ return distributeDataStreamByHashDistributionMode(input, schema, spec);
case RANGE:
- if (equalityFieldIds.isEmpty()) {
- LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and {}=range is not supported yet in flink",
- WRITE_DISTRIBUTION_MODE);
- return input;
- } else {
- LOG.info(
- "Distribute rows by equality fields, because there are equality fields set "
- + "and{}=range is not supported yet in flink",
- WRITE_DISTRIBUTION_MODE);
- return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- }
-
+ return distributeDataStreamByRangeDistributionMode(input, schema, spec, sortOrder);
default:
throw new RuntimeException("Unrecognized " + WRITE_DISTRIBUTION_MODE + ": " + mode);
}
}
+ private DataStream<RowData> distributeDataStreamByNoneDistributionMode(
+ DataStream<RowData> input, Schema iSchema) {
+ if (equalityFieldIds.isEmpty()) {
+ return input;
+ } else {
+ LOG.info("Distribute rows by equality fields, because there are equality fields set");
+ return input.keyBy(new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));
+ }
+ }
+
+ private DataStream<RowData> distributeDataStreamByHashDistributionMode(
+ DataStream<RowData> input, Schema iSchema, PartitionSpec partitionSpec) {
+ if (equalityFieldIds.isEmpty()) {
+ if (partitionSpec.isUnpartitioned()) {
+ LOG.warn(
+ "Fallback to use 'none' distribution mode, because there are no equality fields set "
+ + "and table is unpartitioned");
+ return input;
+ } else {
+ return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
+ }
+ } else {
+ if (partitionSpec.isUnpartitioned()) {
+ LOG.info(
+ "Distribute rows by equality fields, because there are equality fields set "
+ + "and table is unpartitioned");
+ return input.keyBy(new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));
+ } else {
+ for (PartitionField partitionField : partitionSpec.fields()) {
+ Preconditions.checkState(
+ equalityFieldIds.contains(partitionField.sourceId()),
+ "In 'hash' distribution mode with equality fields set, partition field '%s' "
+ + "should be included in equality fields: '%s'",
+ partitionField,
+ equalityFieldColumns);
+ }
+ return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
+ }
+ }
+ }
+
+ private DataStream<RowData> distributeDataStreamByRangeDistributionMode(
+ DataStream<RowData> input,
+ Schema iSchema,
+ PartitionSpec partitionSpec,
+ SortOrder sortOrderParam) {
+
+ int writerParallelism =
Review Comment:
sorry, I wasn't very clear earlier. The v2 sink writer parallelism selection is different than the v1 sink.
```
@Override
public Builder writeParallelism(int newWriteParallelism) {
writeOptions.put(
FlinkWriteOptions.WRITE_PARALLELISM.key(), Integer.toString(newWriteParallelism));
return this;
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Range distribution iceberg sink [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#discussion_r2098767251
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -375,22 +381,43 @@ public Builder flinkConf(ReadableConfig config) {
/**
* Configure the write {@link DistributionMode} that the IcebergSink will use. Currently, flink
- * support {@link DistributionMode#NONE} and {@link DistributionMode#HASH}.
+ * support {@link DistributionMode#NONE} and {@link DistributionMode#HASH} and {@link
+ * DistributionMode#RANGE}
*
* @param mode to specify the write distribution mode.
* @return {@link IcebergSink.Builder} to connect the iceberg table.
*/
@Override
public Builder distributionMode(DistributionMode mode) {
- Preconditions.checkArgument(
- !DistributionMode.RANGE.equals(mode),
- "Flink does not support 'range' write distribution mode now.");
if (mode != null) {
writeOptions.put(FlinkWriteOptions.DISTRIBUTION_MODE.key(), mode.modeName());
}
return this;
}
+ /**
+ * Range distribution needs to collect statistics about data distribution to properly shuffle
+ * the records in relatively balanced way. In general, low cardinality should use {@link
+ * StatisticsType#Map} and high cardinality should use {@link StatisticsType#Sketch} Refer to
+ * {@link StatisticsType} Javadoc for more details.
+ *
+ * <p>Default is {@link StatisticsType#Auto} where initially Map statistics is used. But if
+ * cardinality is higher than the threshold (currently 10K) as defined in {@code
+ * SketchUtil#OPERATOR_SKETCH_SWITCH_THRESHOLD}, statistics collection automatically switches to
+ * the sketch reservoir sampling.
+ *
+ * <p>Explicit set the statistics type if the default behavior doesn't work.
+ *
+ * @param type to specify the statistics type for range distribution.
+ * @return {@link IcebergSink.Builder} to connect the iceberg table.
+ */
+ public IcebergSink.Builder rangeDistributionStatisticsType(StatisticsType type) {
+ if (type != null) {
+ writeOptions.put(FlinkWriteOptions.RANGE_DISTRIBUTION_STATISTICS_TYPE.key(), type.name());
+ }
+ return this;
+ }
+
Review Comment:
it seems that we missed this method
```
public Builder rangeDistributionSortKeyBaseWeight(double weight)
```
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -645,70 +676,121 @@ private DataStream<RowData> distributeDataStream(DataStream<RowData> input) {
DistributionMode mode = flinkWriteConf.distributionMode();
Schema schema = table.schema();
PartitionSpec spec = table.spec();
+ SortOrder sortOrder = table.sortOrder();
+
LOG.info("Write distribution mode is '{}'", mode.modeName());
switch (mode) {
case NONE:
- if (equalityFieldIds.isEmpty()) {
- return input;
- } else {
- LOG.info("Distribute rows by equality fields, because there are equality fields set");
- return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- }
-
+ return distributeDataStreamByNoneDistributionMode(input, schema);
case HASH:
- if (equalityFieldIds.isEmpty()) {
- if (table.spec().isUnpartitioned()) {
- LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and table is unpartitioned");
- return input;
- } else {
- if (BucketPartitionerUtil.hasOneBucketField(spec)) {
- return input.partitionCustom(
- new BucketPartitioner(spec),
- new BucketPartitionKeySelector(spec, schema, flinkRowType));
- } else {
- return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
- }
- }
- } else {
- if (spec.isUnpartitioned()) {
- LOG.info(
- "Distribute rows by equality fields, because there are equality fields set "
- + "and table is unpartitioned");
- return input.keyBy(
- new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- } else {
- for (PartitionField partitionField : spec.fields()) {
- Preconditions.checkState(
- equalityFieldIds.contains(partitionField.sourceId()),
- "In 'hash' distribution mode with equality fields set, partition field '%s' "
- + "should be included in equality fields: '%s'",
- partitionField,
- equalityFieldColumns);
- }
- return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
- }
- }
-
+ return distributeDataStreamByHashDistributionMode(input, schema, spec);
case RANGE:
- if (equalityFieldIds.isEmpty()) {
- LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and {}=range is not supported yet in flink",
- WRITE_DISTRIBUTION_MODE);
- return input;
+ return distributeDataStreamByRangeDistributionMode(input, schema, spec, sortOrder);
+ default:
+ throw new RuntimeException("Unrecognized " + WRITE_DISTRIBUTION_MODE + ": " + mode);
+ }
+ }
+
+ private DataStream<RowData> distributeDataStreamByNoneDistributionMode(
+ DataStream<RowData> input, Schema schema) {
+ if (equalityFieldIds.isEmpty()) {
+ return input;
+ } else {
+ LOG.info("Distribute rows by equality fields, because there are equality fields set");
+ return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
+ }
+ }
+
+ private DataStream<RowData> distributeDataStreamByHashDistributionMode(
+ DataStream<RowData> input, Schema schema, PartitionSpec spec) {
+ if (equalityFieldIds.isEmpty()) {
+ if (table.spec().isUnpartitioned()) {
+ LOG.warn(
+ "Fallback to use 'none' distribution mode, because there are no equality fields set "
+ + "and table is unpartitioned");
+ return input;
+ } else {
+ if (BucketPartitionerUtil.hasOneBucketField(spec)) {
Review Comment:
can you check with `FlinkSink` code again? this code has been removed/reverted there.
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -645,70 +676,121 @@ private DataStream<RowData> distributeDataStream(DataStream<RowData> input) {
DistributionMode mode = flinkWriteConf.distributionMode();
Schema schema = table.schema();
PartitionSpec spec = table.spec();
+ SortOrder sortOrder = table.sortOrder();
+
LOG.info("Write distribution mode is '{}'", mode.modeName());
switch (mode) {
case NONE:
- if (equalityFieldIds.isEmpty()) {
- return input;
- } else {
- LOG.info("Distribute rows by equality fields, because there are equality fields set");
- return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- }
-
+ return distributeDataStreamByNoneDistributionMode(input, schema);
case HASH:
- if (equalityFieldIds.isEmpty()) {
- if (table.spec().isUnpartitioned()) {
- LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and table is unpartitioned");
- return input;
- } else {
- if (BucketPartitionerUtil.hasOneBucketField(spec)) {
- return input.partitionCustom(
- new BucketPartitioner(spec),
- new BucketPartitionKeySelector(spec, schema, flinkRowType));
- } else {
- return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
- }
- }
- } else {
- if (spec.isUnpartitioned()) {
- LOG.info(
- "Distribute rows by equality fields, because there are equality fields set "
- + "and table is unpartitioned");
- return input.keyBy(
- new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- } else {
- for (PartitionField partitionField : spec.fields()) {
- Preconditions.checkState(
- equalityFieldIds.contains(partitionField.sourceId()),
- "In 'hash' distribution mode with equality fields set, partition field '%s' "
- + "should be included in equality fields: '%s'",
- partitionField,
- equalityFieldColumns);
- }
- return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
- }
- }
-
+ return distributeDataStreamByHashDistributionMode(input, schema, spec);
case RANGE:
- if (equalityFieldIds.isEmpty()) {
- LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and {}=range is not supported yet in flink",
- WRITE_DISTRIBUTION_MODE);
- return input;
+ return distributeDataStreamByRangeDistributionMode(input, schema, spec, sortOrder);
+ default:
+ throw new RuntimeException("Unrecognized " + WRITE_DISTRIBUTION_MODE + ": " + mode);
+ }
+ }
+
+ private DataStream<RowData> distributeDataStreamByNoneDistributionMode(
+ DataStream<RowData> input, Schema schema) {
+ if (equalityFieldIds.isEmpty()) {
+ return input;
+ } else {
+ LOG.info("Distribute rows by equality fields, because there are equality fields set");
+ return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
+ }
+ }
+
+ private DataStream<RowData> distributeDataStreamByHashDistributionMode(
+ DataStream<RowData> input, Schema schema, PartitionSpec spec) {
+ if (equalityFieldIds.isEmpty()) {
+ if (table.spec().isUnpartitioned()) {
+ LOG.warn(
+ "Fallback to use 'none' distribution mode, because there are no equality fields set "
+ + "and table is unpartitioned");
+ return input;
+ } else {
+ if (BucketPartitionerUtil.hasOneBucketField(spec)) {
+ return input.partitionCustom(
+ new BucketPartitioner(spec),
+ new BucketPartitionKeySelector(spec, schema, flinkRowType));
} else {
- LOG.info(
- "Distribute rows by equality fields, because there are equality fields set "
- + "and{}=range is not supported yet in flink",
- WRITE_DISTRIBUTION_MODE);
- return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
+ return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
+ }
+ }
+ } else {
+ if (spec.isUnpartitioned()) {
+ LOG.info(
+ "Distribute rows by equality fields, because there are equality fields set "
+ + "and table is unpartitioned");
+ return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
+ } else {
+ for (PartitionField partitionField : spec.fields()) {
+ Preconditions.checkState(
+ equalityFieldIds.contains(partitionField.sourceId()),
+ "In 'hash' distribution mode with equality fields set, partition field '%s' "
+ + "should be included in equality fields: '%s'",
+ partitionField,
+ equalityFieldColumns);
}
+ return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
+ }
+ }
+ }
- default:
- throw new RuntimeException("Unrecognized " + WRITE_DISTRIBUTION_MODE + ": " + mode);
+ private DataStream<RowData> distributeDataStreamByRangeDistributionMode(
+ DataStream<RowData> input, Schema schema, PartitionSpec spec, SortOrder sortOrderParam) {
+
+ int writerParallelism =
+ flinkWriteConf.writeParallelism() == null
+ ? input.getParallelism()
+ : flinkWriteConf.writeParallelism();
+
+ // Ideally, exception should be thrown in the combination of range distribution and
+ // equality fields. Primary key case should use hash distribution mode.
+ // Keep the current behavior of falling back to keyBy for backward compatibility.
+ if (!equalityFieldIds.isEmpty()) {
+ LOG.warn(
+ "Hash distribute rows by equality fields, even though {}=range is set. "
+ + "Range distribution for primary keys are not always safe in "
+ + "Flink streaming writer.",
+ WRITE_DISTRIBUTION_MODE);
+ return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
+ }
+
+ SortOrder sortOrder = sortOrderParam;
+ // range distribute by partition key or sort key if table has an SortOrder
+ Preconditions.checkState(
+ sortOrder.isSorted() || spec.isPartitioned(),
+ "Invalid write distribution mode: range. Need to define sort order or partition spec.");
+ if (sortOrder.isUnsorted()) {
+ sortOrder = Partitioning.sortOrderFor(spec);
+ LOG.info("Construct sort order from partition spec");
}
+
+ LOG.info("Range distribute rows by sort order: {}", sortOrder);
+ StatisticsType statisticsType = flinkWriteConf.rangeDistributionStatisticsType();
+ SingleOutputStreamOperator<StatisticsOrRecord> shuffleStream =
+ input
+ .transform(
+ operatorName("range-shuffle"),
+ TypeInformation.of(StatisticsOrRecord.class),
+ new DataStatisticsOperatorFactory(
+ schema,
+ sortOrder,
+ writerParallelism,
+ statisticsType,
+ flinkWriteConf.rangeDistributionSortKeyBaseWeight()))
+ // Set the parallelism same as input operator to encourage chaining
+ .setParallelism(input.getParallelism());
+ if (uidSuffix != null) {
+ shuffleStream = shuffleStream.uid("shuffle-" + uidSuffix);
+ }
+
+ return shuffleStream
+ .partitionCustom(new RangePartitioner(schema, sortOrder), r -> r)
+ .filter(StatisticsOrRecord::hasRecord)
Review Comment:
there has been changes from when this PR was initially created. please re-sync with `FlinkSink`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: port range distribution to v2 iceberg sink [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#discussion_r2101587781
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -645,72 +711,135 @@ private DataStream<RowData> distributeDataStream(DataStream<RowData> input) {
DistributionMode mode = flinkWriteConf.distributionMode();
Schema schema = table.schema();
PartitionSpec spec = table.spec();
+ SortOrder sortOrder = table.sortOrder();
+
LOG.info("Write distribution mode is '{}'", mode.modeName());
switch (mode) {
case NONE:
- if (equalityFieldIds.isEmpty()) {
- return input;
- } else {
- LOG.info("Distribute rows by equality fields, because there are equality fields set");
- return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- }
-
+ return distributeDataStreamByNoneDistributionMode(input, schema);
case HASH:
- if (equalityFieldIds.isEmpty()) {
- if (table.spec().isUnpartitioned()) {
- LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and table is unpartitioned");
- return input;
- } else {
- if (BucketPartitionerUtil.hasOneBucketField(spec)) {
- return input.partitionCustom(
- new BucketPartitioner(spec),
- new BucketPartitionKeySelector(spec, schema, flinkRowType));
- } else {
- return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
- }
- }
- } else {
- if (spec.isUnpartitioned()) {
- LOG.info(
- "Distribute rows by equality fields, because there are equality fields set "
- + "and table is unpartitioned");
- return input.keyBy(
- new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- } else {
- for (PartitionField partitionField : spec.fields()) {
- Preconditions.checkState(
- equalityFieldIds.contains(partitionField.sourceId()),
- "In 'hash' distribution mode with equality fields set, partition field '%s' "
- + "should be included in equality fields: '%s'",
- partitionField,
- equalityFieldColumns);
- }
- return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
- }
- }
-
+ return distributeDataStreamByHashDistributionMode(input, schema, spec);
case RANGE:
- if (equalityFieldIds.isEmpty()) {
- LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and {}=range is not supported yet in flink",
- WRITE_DISTRIBUTION_MODE);
- return input;
- } else {
- LOG.info(
- "Distribute rows by equality fields, because there are equality fields set "
- + "and{}=range is not supported yet in flink",
- WRITE_DISTRIBUTION_MODE);
- return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- }
-
+ return distributeDataStreamByRangeDistributionMode(input, schema, spec, sortOrder);
default:
throw new RuntimeException("Unrecognized " + WRITE_DISTRIBUTION_MODE + ": " + mode);
}
}
+ private DataStream<RowData> distributeDataStreamByNoneDistributionMode(
+ DataStream<RowData> input, Schema iSchema) {
+ if (equalityFieldIds.isEmpty()) {
+ return input;
+ } else {
+ LOG.info("Distribute rows by equality fields, because there are equality fields set");
+ return input.keyBy(new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));
+ }
+ }
+
+ private DataStream<RowData> distributeDataStreamByHashDistributionMode(
+ DataStream<RowData> input, Schema iSchema, PartitionSpec partitionSpec) {
+ if (equalityFieldIds.isEmpty()) {
+ if (partitionSpec.isUnpartitioned()) {
+ LOG.warn(
+ "Fallback to use 'none' distribution mode, because there are no equality fields set "
+ + "and table is unpartitioned");
+ return input;
+ } else {
+ return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
+ }
+ } else {
+ if (partitionSpec.isUnpartitioned()) {
+ LOG.info(
+ "Distribute rows by equality fields, because there are equality fields set "
+ + "and table is unpartitioned");
+ return input.keyBy(new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));
+ } else {
+ for (PartitionField partitionField : partitionSpec.fields()) {
+ Preconditions.checkState(
+ equalityFieldIds.contains(partitionField.sourceId()),
+ "In 'hash' distribution mode with equality fields set, partition field '%s' "
+ + "should be included in equality fields: '%s'",
+ partitionField,
+ equalityFieldColumns);
+ }
+ return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
+ }
+ }
+ }
+
+ private DataStream<RowData> distributeDataStreamByRangeDistributionMode(
+ DataStream<RowData> input,
+ Schema iSchema,
+ PartitionSpec partitionSpec,
+ SortOrder sortOrderParam) {
+
+ int writerParallelism =
Review Comment:
this logic should be applied to the writer parallelism for v2 sink.
if write parallelism is not configured, the v1 sink default the writer parallelism to the input parallelism to promote chaining. Want to confirm if that is the case for v2 sink? from reading the code, I thought v2 sink will default the writer parallelism to default job parallelism?
```
// Note that IcebergSink internally consists o multiple operators (like writer, committer,
// aggregator).
// The following parallelism will be propagated to all of the above operators.
if (sink.flinkWriteConf.writeParallelism() != null) {
rowDataDataStreamSink.setParallelism(sink.flinkWriteConf.writeParallelism());
}
```
technically, if this is a behavior change problem for v2 sink, it is not caused by this PR. but it is critical that the same writer parallelism is used by the shuffle operator to properly range partition the data to downstream writer tasks. That is why in the v1 `FlinkSink`, you can see `writerParallelism` is computed once and pass to two methods.
```
int writerParallelism =
flinkWriteConf.writeParallelism() == null
? rowDataInput.getParallelism()
: flinkWriteConf.writeParallelism();
// Distribute the records from input data stream based on the write.distribution-mode and
// equality fields.
DataStream<RowData> distributeStream =
distributeDataStream(rowDataInput, equalityFieldIds, flinkRowType, writerParallelism);
// Add parallel writers that append rows to files
SingleOutputStreamOperator<FlinkWriteResult> writerStream =
appendWriter(distributeStream, flinkRowType, equalityFieldIds, writerParallelism);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Range distribution iceberg sink [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#issuecomment-2642745078
@pvary Can you take a look as well?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: port range distribution to v2 iceberg sink [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#discussion_r2101587781
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -645,72 +711,135 @@ private DataStream<RowData> distributeDataStream(DataStream<RowData> input) {
DistributionMode mode = flinkWriteConf.distributionMode();
Schema schema = table.schema();
PartitionSpec spec = table.spec();
+ SortOrder sortOrder = table.sortOrder();
+
LOG.info("Write distribution mode is '{}'", mode.modeName());
switch (mode) {
case NONE:
- if (equalityFieldIds.isEmpty()) {
- return input;
- } else {
- LOG.info("Distribute rows by equality fields, because there are equality fields set");
- return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- }
-
+ return distributeDataStreamByNoneDistributionMode(input, schema);
case HASH:
- if (equalityFieldIds.isEmpty()) {
- if (table.spec().isUnpartitioned()) {
- LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and table is unpartitioned");
- return input;
- } else {
- if (BucketPartitionerUtil.hasOneBucketField(spec)) {
- return input.partitionCustom(
- new BucketPartitioner(spec),
- new BucketPartitionKeySelector(spec, schema, flinkRowType));
- } else {
- return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
- }
- }
- } else {
- if (spec.isUnpartitioned()) {
- LOG.info(
- "Distribute rows by equality fields, because there are equality fields set "
- + "and table is unpartitioned");
- return input.keyBy(
- new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- } else {
- for (PartitionField partitionField : spec.fields()) {
- Preconditions.checkState(
- equalityFieldIds.contains(partitionField.sourceId()),
- "In 'hash' distribution mode with equality fields set, partition field '%s' "
- + "should be included in equality fields: '%s'",
- partitionField,
- equalityFieldColumns);
- }
- return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
- }
- }
-
+ return distributeDataStreamByHashDistributionMode(input, schema, spec);
case RANGE:
- if (equalityFieldIds.isEmpty()) {
- LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and {}=range is not supported yet in flink",
- WRITE_DISTRIBUTION_MODE);
- return input;
- } else {
- LOG.info(
- "Distribute rows by equality fields, because there are equality fields set "
- + "and{}=range is not supported yet in flink",
- WRITE_DISTRIBUTION_MODE);
- return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- }
-
+ return distributeDataStreamByRangeDistributionMode(input, schema, spec, sortOrder);
default:
throw new RuntimeException("Unrecognized " + WRITE_DISTRIBUTION_MODE + ": " + mode);
}
}
+ private DataStream<RowData> distributeDataStreamByNoneDistributionMode(
+ DataStream<RowData> input, Schema iSchema) {
+ if (equalityFieldIds.isEmpty()) {
+ return input;
+ } else {
+ LOG.info("Distribute rows by equality fields, because there are equality fields set");
+ return input.keyBy(new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));
+ }
+ }
+
+ private DataStream<RowData> distributeDataStreamByHashDistributionMode(
+ DataStream<RowData> input, Schema iSchema, PartitionSpec partitionSpec) {
+ if (equalityFieldIds.isEmpty()) {
+ if (partitionSpec.isUnpartitioned()) {
+ LOG.warn(
+ "Fallback to use 'none' distribution mode, because there are no equality fields set "
+ + "and table is unpartitioned");
+ return input;
+ } else {
+ return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
+ }
+ } else {
+ if (partitionSpec.isUnpartitioned()) {
+ LOG.info(
+ "Distribute rows by equality fields, because there are equality fields set "
+ + "and table is unpartitioned");
+ return input.keyBy(new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));
+ } else {
+ for (PartitionField partitionField : partitionSpec.fields()) {
+ Preconditions.checkState(
+ equalityFieldIds.contains(partitionField.sourceId()),
+ "In 'hash' distribution mode with equality fields set, partition field '%s' "
+ + "should be included in equality fields: '%s'",
+ partitionField,
+ equalityFieldColumns);
+ }
+ return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
+ }
+ }
+ }
+
+ private DataStream<RowData> distributeDataStreamByRangeDistributionMode(
+ DataStream<RowData> input,
+ Schema iSchema,
+ PartitionSpec partitionSpec,
+ SortOrder sortOrderParam) {
+
+ int writerParallelism =
Review Comment:
this logic should be applied to the writer parallelism for v2 sink.
if write parallelism is not configured, the v1 sink default the writer parallelism to the input parallelism to promote chaining. Want to confirm if that is the case for v2 sink? from reading the code, I thought v2 sink will default the writer parallelism to default job parallelism?
```
// Note that IcebergSink internally consists o multiple operators (like writer, committer,
// aggregator).
// The following parallelism will be propagated to all of the above operators.
if (sink.flinkWriteConf.writeParallelism() != null) {
rowDataDataStreamSink.setParallelism(sink.flinkWriteConf.writeParallelism());
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: port range distribution to v2 iceberg sink [iceberg]
Posted by "rodmeneses (via GitHub)" <gi...@apache.org>.
rodmeneses commented on code in PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#discussion_r2112131384
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -645,72 +711,135 @@ private DataStream<RowData> distributeDataStream(DataStream<RowData> input) {
DistributionMode mode = flinkWriteConf.distributionMode();
Schema schema = table.schema();
PartitionSpec spec = table.spec();
+ SortOrder sortOrder = table.sortOrder();
+
LOG.info("Write distribution mode is '{}'", mode.modeName());
switch (mode) {
case NONE:
- if (equalityFieldIds.isEmpty()) {
- return input;
- } else {
- LOG.info("Distribute rows by equality fields, because there are equality fields set");
- return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- }
-
+ return distributeDataStreamByNoneDistributionMode(input, schema);
case HASH:
- if (equalityFieldIds.isEmpty()) {
- if (table.spec().isUnpartitioned()) {
- LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and table is unpartitioned");
- return input;
- } else {
- if (BucketPartitionerUtil.hasOneBucketField(spec)) {
- return input.partitionCustom(
- new BucketPartitioner(spec),
- new BucketPartitionKeySelector(spec, schema, flinkRowType));
- } else {
- return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
- }
- }
- } else {
- if (spec.isUnpartitioned()) {
- LOG.info(
- "Distribute rows by equality fields, because there are equality fields set "
- + "and table is unpartitioned");
- return input.keyBy(
- new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- } else {
- for (PartitionField partitionField : spec.fields()) {
- Preconditions.checkState(
- equalityFieldIds.contains(partitionField.sourceId()),
- "In 'hash' distribution mode with equality fields set, partition field '%s' "
- + "should be included in equality fields: '%s'",
- partitionField,
- equalityFieldColumns);
- }
- return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
- }
- }
-
+ return distributeDataStreamByHashDistributionMode(input, schema, spec);
case RANGE:
- if (equalityFieldIds.isEmpty()) {
- LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and {}=range is not supported yet in flink",
- WRITE_DISTRIBUTION_MODE);
- return input;
- } else {
- LOG.info(
- "Distribute rows by equality fields, because there are equality fields set "
- + "and{}=range is not supported yet in flink",
- WRITE_DISTRIBUTION_MODE);
- return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- }
-
+ return distributeDataStreamByRangeDistributionMode(input, schema, spec, sortOrder);
default:
throw new RuntimeException("Unrecognized " + WRITE_DISTRIBUTION_MODE + ": " + mode);
}
}
+ private DataStream<RowData> distributeDataStreamByNoneDistributionMode(
+ DataStream<RowData> input, Schema iSchema) {
+ if (equalityFieldIds.isEmpty()) {
+ return input;
+ } else {
+ LOG.info("Distribute rows by equality fields, because there are equality fields set");
+ return input.keyBy(new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));
+ }
+ }
+
+ private DataStream<RowData> distributeDataStreamByHashDistributionMode(
+ DataStream<RowData> input, Schema iSchema, PartitionSpec partitionSpec) {
+ if (equalityFieldIds.isEmpty()) {
+ if (partitionSpec.isUnpartitioned()) {
+ LOG.warn(
+ "Fallback to use 'none' distribution mode, because there are no equality fields set "
+ + "and table is unpartitioned");
+ return input;
+ } else {
+ return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
+ }
+ } else {
+ if (partitionSpec.isUnpartitioned()) {
+ LOG.info(
+ "Distribute rows by equality fields, because there are equality fields set "
+ + "and table is unpartitioned");
+ return input.keyBy(new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));
+ } else {
+ for (PartitionField partitionField : partitionSpec.fields()) {
+ Preconditions.checkState(
+ equalityFieldIds.contains(partitionField.sourceId()),
+ "In 'hash' distribution mode with equality fields set, partition field '%s' "
+ + "should be included in equality fields: '%s'",
+ partitionField,
+ equalityFieldColumns);
+ }
+ return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
+ }
+ }
+ }
+
+ private DataStream<RowData> distributeDataStreamByRangeDistributionMode(
+ DataStream<RowData> input,
+ Schema iSchema,
+ PartitionSpec partitionSpec,
+ SortOrder sortOrderParam) {
+
+ int writerParallelism =
Review Comment:
Hi @stevenzwu . Thanks for the clarification. I think I know what you mean. On FlinkSink, even without taking about RANGE distribution, the parallelism of the v1 sink by default will be the same as the input source parallelism.
This is a good approach, because it encourage chaining.
However, we dont have that logic on the IcebergSink v2. I think we should have the same logic there. I could do that in another PR and the follow up with this one. What do you think? Thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: port range distribution to v2 iceberg sink [iceberg]
Posted by "rodmeneses (via GitHub)" <gi...@apache.org>.
rodmeneses commented on code in PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#discussion_r2121533610
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -645,72 +711,135 @@ private DataStream<RowData> distributeDataStream(DataStream<RowData> input) {
DistributionMode mode = flinkWriteConf.distributionMode();
Schema schema = table.schema();
PartitionSpec spec = table.spec();
+ SortOrder sortOrder = table.sortOrder();
+
LOG.info("Write distribution mode is '{}'", mode.modeName());
switch (mode) {
case NONE:
- if (equalityFieldIds.isEmpty()) {
- return input;
- } else {
- LOG.info("Distribute rows by equality fields, because there are equality fields set");
- return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- }
-
+ return distributeDataStreamByNoneDistributionMode(input, schema);
case HASH:
- if (equalityFieldIds.isEmpty()) {
- if (table.spec().isUnpartitioned()) {
- LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and table is unpartitioned");
- return input;
- } else {
- if (BucketPartitionerUtil.hasOneBucketField(spec)) {
- return input.partitionCustom(
- new BucketPartitioner(spec),
- new BucketPartitionKeySelector(spec, schema, flinkRowType));
- } else {
- return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
- }
- }
- } else {
- if (spec.isUnpartitioned()) {
- LOG.info(
- "Distribute rows by equality fields, because there are equality fields set "
- + "and table is unpartitioned");
- return input.keyBy(
- new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- } else {
- for (PartitionField partitionField : spec.fields()) {
- Preconditions.checkState(
- equalityFieldIds.contains(partitionField.sourceId()),
- "In 'hash' distribution mode with equality fields set, partition field '%s' "
- + "should be included in equality fields: '%s'",
- partitionField,
- equalityFieldColumns);
- }
- return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
- }
- }
-
+ return distributeDataStreamByHashDistributionMode(input, schema, spec);
case RANGE:
- if (equalityFieldIds.isEmpty()) {
- LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and {}=range is not supported yet in flink",
- WRITE_DISTRIBUTION_MODE);
- return input;
- } else {
- LOG.info(
- "Distribute rows by equality fields, because there are equality fields set "
- + "and{}=range is not supported yet in flink",
- WRITE_DISTRIBUTION_MODE);
- return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- }
-
+ return distributeDataStreamByRangeDistributionMode(input, schema, spec, sortOrder);
default:
throw new RuntimeException("Unrecognized " + WRITE_DISTRIBUTION_MODE + ": " + mode);
}
}
+ private DataStream<RowData> distributeDataStreamByNoneDistributionMode(
+ DataStream<RowData> input, Schema iSchema) {
+ if (equalityFieldIds.isEmpty()) {
+ return input;
+ } else {
+ LOG.info("Distribute rows by equality fields, because there are equality fields set");
+ return input.keyBy(new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));
+ }
+ }
+
+ private DataStream<RowData> distributeDataStreamByHashDistributionMode(
+ DataStream<RowData> input, Schema iSchema, PartitionSpec partitionSpec) {
+ if (equalityFieldIds.isEmpty()) {
+ if (partitionSpec.isUnpartitioned()) {
+ LOG.warn(
+ "Fallback to use 'none' distribution mode, because there are no equality fields set "
+ + "and table is unpartitioned");
+ return input;
+ } else {
+ return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
+ }
+ } else {
+ if (partitionSpec.isUnpartitioned()) {
+ LOG.info(
+ "Distribute rows by equality fields, because there are equality fields set "
+ + "and table is unpartitioned");
+ return input.keyBy(new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));
+ } else {
+ for (PartitionField partitionField : partitionSpec.fields()) {
+ Preconditions.checkState(
+ equalityFieldIds.contains(partitionField.sourceId()),
+ "In 'hash' distribution mode with equality fields set, partition field '%s' "
+ + "should be included in equality fields: '%s'",
+ partitionField,
+ equalityFieldColumns);
+ }
+ return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
+ }
+ }
+ }
+
+ private DataStream<RowData> distributeDataStreamByRangeDistributionMode(
+ DataStream<RowData> input,
+ Schema iSchema,
+ PartitionSpec partitionSpec,
+ SortOrder sortOrderParam) {
+
+ int writerParallelism =
Review Comment:
will do!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: port range distribution to v2 iceberg sink [iceberg]
Posted by "rodmeneses (via GitHub)" <gi...@apache.org>.
rodmeneses commented on code in PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#discussion_r2121884952
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -645,72 +711,135 @@ private DataStream<RowData> distributeDataStream(DataStream<RowData> input) {
DistributionMode mode = flinkWriteConf.distributionMode();
Schema schema = table.schema();
PartitionSpec spec = table.spec();
+ SortOrder sortOrder = table.sortOrder();
+
LOG.info("Write distribution mode is '{}'", mode.modeName());
switch (mode) {
case NONE:
- if (equalityFieldIds.isEmpty()) {
- return input;
- } else {
- LOG.info("Distribute rows by equality fields, because there are equality fields set");
- return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- }
-
+ return distributeDataStreamByNoneDistributionMode(input, schema);
case HASH:
- if (equalityFieldIds.isEmpty()) {
- if (table.spec().isUnpartitioned()) {
- LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and table is unpartitioned");
- return input;
- } else {
- if (BucketPartitionerUtil.hasOneBucketField(spec)) {
- return input.partitionCustom(
- new BucketPartitioner(spec),
- new BucketPartitionKeySelector(spec, schema, flinkRowType));
- } else {
- return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
- }
- }
- } else {
- if (spec.isUnpartitioned()) {
- LOG.info(
- "Distribute rows by equality fields, because there are equality fields set "
- + "and table is unpartitioned");
- return input.keyBy(
- new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- } else {
- for (PartitionField partitionField : spec.fields()) {
- Preconditions.checkState(
- equalityFieldIds.contains(partitionField.sourceId()),
- "In 'hash' distribution mode with equality fields set, partition field '%s' "
- + "should be included in equality fields: '%s'",
- partitionField,
- equalityFieldColumns);
- }
- return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
- }
- }
-
+ return distributeDataStreamByHashDistributionMode(input, schema, spec);
case RANGE:
- if (equalityFieldIds.isEmpty()) {
- LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and {}=range is not supported yet in flink",
- WRITE_DISTRIBUTION_MODE);
- return input;
- } else {
- LOG.info(
- "Distribute rows by equality fields, because there are equality fields set "
- + "and{}=range is not supported yet in flink",
- WRITE_DISTRIBUTION_MODE);
- return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- }
-
+ return distributeDataStreamByRangeDistributionMode(input, schema, spec, sortOrder);
default:
throw new RuntimeException("Unrecognized " + WRITE_DISTRIBUTION_MODE + ": " + mode);
}
}
+ private DataStream<RowData> distributeDataStreamByNoneDistributionMode(
+ DataStream<RowData> input, Schema iSchema) {
+ if (equalityFieldIds.isEmpty()) {
+ return input;
+ } else {
+ LOG.info("Distribute rows by equality fields, because there are equality fields set");
+ return input.keyBy(new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));
+ }
+ }
+
+ private DataStream<RowData> distributeDataStreamByHashDistributionMode(
+ DataStream<RowData> input, Schema iSchema, PartitionSpec partitionSpec) {
+ if (equalityFieldIds.isEmpty()) {
+ if (partitionSpec.isUnpartitioned()) {
+ LOG.warn(
+ "Fallback to use 'none' distribution mode, because there are no equality fields set "
+ + "and table is unpartitioned");
+ return input;
+ } else {
+ return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
+ }
+ } else {
+ if (partitionSpec.isUnpartitioned()) {
+ LOG.info(
+ "Distribute rows by equality fields, because there are equality fields set "
+ + "and table is unpartitioned");
+ return input.keyBy(new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));
+ } else {
+ for (PartitionField partitionField : partitionSpec.fields()) {
+ Preconditions.checkState(
+ equalityFieldIds.contains(partitionField.sourceId()),
+ "In 'hash' distribution mode with equality fields set, partition field '%s' "
+ + "should be included in equality fields: '%s'",
+ partitionField,
+ equalityFieldColumns);
+ }
+ return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
+ }
+ }
+ }
+
+ private DataStream<RowData> distributeDataStreamByRangeDistributionMode(
+ DataStream<RowData> input,
+ Schema iSchema,
+ PartitionSpec partitionSpec,
+ SortOrder sortOrderParam) {
+
+ int writerParallelism =
Review Comment:
Hi @stevenzwu I have rebased and this is ready! Thanks
Tagging @mxm and @pvary as well
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: port range distribution to v2 iceberg sink [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#issuecomment-2937104181
thanks @rodmeneses for the contribution and @mxm @pvary for the review
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Range distribution iceberg sink [iceberg]
Posted by "rodmeneses (via GitHub)" <gi...@apache.org>.
rodmeneses commented on code in PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#discussion_r2098754472
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -645,70 +676,121 @@ private DataStream<RowData> distributeDataStream(DataStream<RowData> input) {
DistributionMode mode = flinkWriteConf.distributionMode();
Schema schema = table.schema();
PartitionSpec spec = table.spec();
+ SortOrder sortOrder = table.sortOrder();
+
LOG.info("Write distribution mode is '{}'", mode.modeName());
switch (mode) {
case NONE:
- if (equalityFieldIds.isEmpty()) {
- return input;
- } else {
- LOG.info("Distribute rows by equality fields, because there are equality fields set");
- return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- }
-
+ return distributeDataStreamByNoneDistributionMode(input, schema);
case HASH:
- if (equalityFieldIds.isEmpty()) {
- if (table.spec().isUnpartitioned()) {
- LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and table is unpartitioned");
- return input;
- } else {
- if (BucketPartitionerUtil.hasOneBucketField(spec)) {
- return input.partitionCustom(
- new BucketPartitioner(spec),
- new BucketPartitionKeySelector(spec, schema, flinkRowType));
- } else {
- return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
- }
- }
- } else {
- if (spec.isUnpartitioned()) {
- LOG.info(
- "Distribute rows by equality fields, because there are equality fields set "
- + "and table is unpartitioned");
- return input.keyBy(
- new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
- } else {
- for (PartitionField partitionField : spec.fields()) {
- Preconditions.checkState(
- equalityFieldIds.contains(partitionField.sourceId()),
- "In 'hash' distribution mode with equality fields set, partition field '%s' "
- + "should be included in equality fields: '%s'",
- partitionField,
- equalityFieldColumns);
- }
- return input.keyBy(new PartitionKeySelector(spec, schema, flinkRowType));
- }
- }
-
+ return distributeDataStreamByHashDistributionMode(input, schema, spec);
case RANGE:
- if (equalityFieldIds.isEmpty()) {
- LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and {}=range is not supported yet in flink",
- WRITE_DISTRIBUTION_MODE);
- return input;
+ return distributeDataStreamByRangeDistributionMode(input, schema, spec, sortOrder);
+ default:
+ throw new RuntimeException("Unrecognized " + WRITE_DISTRIBUTION_MODE + ": " + mode);
+ }
+ }
+
+ private DataStream<RowData> distributeDataStreamByNoneDistributionMode(
Review Comment:
Breaking `distributeDataStream` into smaller functions, each one for a given distributionMode. due to:
```
Cyclomatic Complexity is 13 (max allowed is 12). [CyclomaticComplexity]
```
This way, it is also clear what each distributionMode needs as function parameters, i.e:
1. `distributeDataStreamByNoneDistributionMode -> (DataStream<RowData> input, Schema schema) `
2. `distributeDataStreamByHashDistributionMode -> (DataStream<RowData> input, Schema schema, PartitionSpec spec)`
3. `distributeDataStreamByRangeDistributionMode` ->
DataStream<RowData> input, Schema schema, PartitionSpec spec, SortOrder sortOrderParam`
Which brings nice information about what each distributionMode needs for its internal calculation
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Range distribution iceberg sink [iceberg]
Posted by "rodmeneses (via GitHub)" <gi...@apache.org>.
rodmeneses commented on code in PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#discussion_r2098984273
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSink.java:
##########
@@ -140,6 +140,13 @@ void testWriteRowWithTableSchema() throws Exception {
testWriteRow(SimpleDataUtil.FLINK_SCHEMA, DistributionMode.NONE);
}
+ @TestTemplate
+ void testWriteRowWithTableSchemaRANGE() throws Exception {
+ if (partitioned) {
+ testWriteRow(SimpleDataUtil.FLINK_SCHEMA, DistributionMode.RANGE);
Review Comment:
not in this same unit test, as `env.execute` traverse the graph.
I did implement it in another unit test. But it is hacky as there's no type safe way to get an instance of the underlying `Partitioner` that is wrapped around a `CustomPartitionerWrapper`. However, with reflection magic, it is possible. I did see precedent of other unit tests leveraging Java Reflection. So, I think this is helpful, but please let me know your thoughts @pvary
```java
@TestTemplate
void testRangePartitionedIsSet() throws Exception {
if (partitioned) {
List<Row> rows = createRows("");
DataStream<Row> dataStream =
env.addSource(createBoundedSource(rows), ROW_TYPE_INFO).uid("mySourceId");
IcebergSink.forRow(dataStream, SimpleDataUtil.FLINK_SCHEMA)
.table(table)
.tableLoader(tableLoader)
.tableSchema(SimpleDataUtil.FLINK_SCHEMA)
.writeParallelism(parallelism)
.distributionMode(DistributionMode.RANGE)
.append();
boolean found = false;
for (StreamNode node : env.getStreamGraph().getStreamNodes()) {
for (StreamEdge edge : node.getOutEdges()) {
if (edge.getPartitioner() instanceof CustomPartitionerWrapper) {
CustomPartitionerWrapper wrapper = (CustomPartitionerWrapper) edge.getPartitioner();
try {
Field privatePartitionerField = CustomPartitionerWrapper.class.getDeclaredField("partitioner");
privatePartitionerField.setAccessible(true);
Object wrappedPartitionerObject = privatePartitionerField.get(wrapper);
if (wrappedPartitionerObject instanceof RangePartitioner) {
found = true;
}
} catch (Exception e) {
throw new RuntimeException(e);
}
}
}
}
assertThat(found)
.as("RangePartitioner was found")
.isTrue();
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Range distribution iceberg sink [iceberg]
Posted by "rodmeneses (via GitHub)" <gi...@apache.org>.
rodmeneses commented on code in PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#discussion_r2099051067
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSink.java:
##########
@@ -145,57 +145,6 @@ void testWriteRowWithTableSchema() throws Exception {
testWriteRow(SimpleDataUtil.FLINK_SCHEMA, DistributionMode.NONE);
}
- @TestTemplate
- void testJobNoneDistributeMode() throws Exception {
Review Comment:
All these methods were incorrectly copied here in previous commit and they are duplicates of the ones on `TestFlinkIcebergSinkV2DistributionMode`
So I'm removing them
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Range distribution iceberg sink [iceberg]
Posted by "rodmeneses (via GitHub)" <gi...@apache.org>.
rodmeneses commented on PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#issuecomment-2887181029
Hi @stevenzwu thanks for reopening this. I will try to finish it this coming week. I think it needs only to port some fixes recently made in the FlinkSink RANGE distribution mode, as well as address some reviews.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Range distribution iceberg sink [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12071:
URL: https://github.com/apache/iceberg/pull/12071#discussion_r1946490418
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -692,20 +729,51 @@ private DataStream<RowData> distributeDataStream(DataStream<RowData> input) {
}
case RANGE:
- if (equalityFieldIds.isEmpty()) {
+ // Ideally, exception should be thrown in the combination of range distribution and
+ // equality fields. Primary key case should use hash distribution mode.
+ // Keep the current behavior of falling back to keyBy for backward compatibility.
+ if (!equalityFieldIds.isEmpty()) {
LOG.warn(
- "Fallback to use 'none' distribution mode, because there are no equality fields set "
- + "and {}=range is not supported yet in flink",
- WRITE_DISTRIBUTION_MODE);
- return input;
- } else {
- LOG.info(
- "Distribute rows by equality fields, because there are equality fields set "
- + "and{}=range is not supported yet in flink",
+ "Hash distribute rows by equality fields, even though {}=range is set. "
+ + "Range distribution for primary keys are not always safe in "
+ + "Flink streaming writer.",
WRITE_DISTRIBUTION_MODE);
return input.keyBy(new EqualityFieldKeySelector(schema, flinkRowType, equalityFieldIds));
}
+ // range distribute by partition key or sort key if table has an SortOrder
+ Preconditions.checkState(
+ sortOrder.isSorted() || spec.isPartitioned(),
+ "Invalid write distribution mode: range. Need to define sort order or partition spec.");
+ if (sortOrder.isUnsorted()) {
+ sortOrder = Partitioning.sortOrderFor(spec);
+ LOG.info("Construct sort order from partition spec");
+ }
+
+ LOG.info("Range distribute rows by sort order: {}", sortOrder);
+ StatisticsType statisticsType = flinkWriteConf.rangeDistributionStatisticsType();
+ SingleOutputStreamOperator<StatisticsOrRecord> shuffleStream =
+ input
+ .transform(
+ operatorName("range-shuffle"),
+ TypeInformation.of(StatisticsOrRecord.class),
+ new DataStatisticsOperatorFactory(
+ schema,
+ sortOrder,
+ writerParallelism,
+ statisticsType,
+ flinkWriteConf.rangeDistributionSortKeyBaseWeight()))
+ // Set the parallelism same as input operator to encourage chaining
+ .setParallelism(input.getParallelism());
Review Comment:
What happens if "encouraging" does not help?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org