Flink: use correct metric config for position deletes #6313

chenjunjiedada · 2022-11-30T02:23:27Z

FlinkAppenderFactory uses the wrong metric configs for position deletes, leading to ineffective filtering. Refactoring the whole write path to FileWriterFactory could fix this while we are using FileAppenderFactory for a long time. Not sure what other benefits can bring if we change that. So here is a simple fix.

chenjunjiedada · 2022-11-30T02:24:13Z

build.gradle

      oldGroup = project.group
      oldName = project.name
-      oldVersion = "1.1.0"
+      oldVersion = "1.0.0"


Just for build, will change back if the master pass the build.

chenjunjiedada · 2022-11-30T02:25:27Z

@stevenzwu @hililiwei You guys may have an interest in this.

stevenzwu · 2022-11-30T04:15:09Z

core/src/main/java/org/apache/iceberg/MetricsConfig.java

+   *
+   * @param props table configuration
+   */
+  public static MetricsConfig forPositionDelete(Map<String, String> props) {


should we use the method from line 91?

public static MetricsConfig forPositionDelete(Table table) {

Flink write properties allows override of table properties for write (e.g. compression configs). Do we have such use cases for metrics config?

I am asking because the current fromProperties method (for data files) is deprecated in favor of the method forTable. Looks like deprecation is from PR #2240 by @szehon-ho .

/** * Creates a metrics config from table configuration. * * @param props table configuration * @deprecated use {@link MetricsConfig#forTable(Table)} */ @Deprecated public static MetricsConfig fromProperties(Map<String, String> props) {

At least, we can avoid code duplication by extracting common code btw these two methods, if we do need this new variant.

Extraction done.

@chenjunjiedada what do you think of using the existing API? I know it would require changes in FlinkAppenderFactory to pass in the Table object.

public static MetricsConfig forPositionDelete(Table table) {

The benefit is that position delete file will get the consistent behavior as data file. unless we are saying position delete has its own schema, shared code path doesn't really make sense.

private static MetricsConfig from(Map<String, String> props, Schema schema, SortOrder order) { ... }

Actually, positions delete does NOT share the same behavior as data files, SortedPosDeleteWriter sorts the content.

What do you mean by has its own schema? Position deletes can have two schemas, one is [path, pos] and another is [path, pos, row]. Currently, it writes [path, pos] in BaseTaskWriter.

@stevenzwu I recheck the metric config, the schema and sort order are for the data row. Change to use forTable now.

@chenjunjiedada hmm. I think we should revert the last commit. if we are going to pass in the Table object, it should be non-null. Then we can consistently use forTable everywhere.

I would also be open with the forProperties approach, as you were saying that schema and sort order are for data rows/files. I had some reservations because MetricConfig deprecated forProperties for data rows/files. It is a little weird to introduce it back for position delete. Flink can use forPositionDelete(Table table) API by passing a valid SerializableTable object to FlinkAppenderFactory. With Table object, we can remove the Schema and PartitionSpec from the current FlinkAppenderFactory constructor

BTW, Anton had a previous attempt for moving Flink to SerializableTable for Flink source/reader. #2987.

We can revive that effort as a separate PR? this can be merged after that.

Let me take a look at that one fisrtly.

@stevenzwu , That one (#6407) is for source. This PR is for the sink side. Let me take a look at the sink as well. Maybe eventually we will change back to adopt FileWriterFactory.

stevenzwu · 2022-11-30T04:21:27Z

core/src/main/java/org/apache/iceberg/MetricsConfig.java

+    columnModes.put(MetadataColumns.DELETE_FILE_PATH.name(), MetricsModes.Full.get());
+    columnModes.put(MetadataColumns.DELETE_FILE_POS.name(), MetricsModes.Full.get());
+
+    MetricsConfig tableConfig = from(props, null, null);


since this is not forTable method, it shouldn't be called tableConfig.

hililiwei

I tend to use this simple fix first.

chenjunjiedada · 2022-12-06T03:50:14Z

flink/v1.15/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkAppenderFactory.java


-    MetricsConfig metricsConfig = MetricsConfig.fromProperties(props);
+    MetricsConfig metricsConfig =
+        table != null ? MetricsConfig.forTable(table) : MetricsConfig.fromProperties(props);


The data row in equality delete should share the same metrics config as the table, so I change this as well.

this actually makes a stronger case that we should pass in a valid Table object to the FlinkAppenderFactory as I mentioned in the other comment. we should also use forTable for position delete although it doesn't need schema or SortOrder as you said.

stevenzwu · 2022-12-13T02:50:23Z

flink/v1.16/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkAppenderFactory.java

      EncryptedOutputFile outputFile, FileFormat format, StructLike partition) {
-    MetricsConfig metricsConfig = MetricsConfig.fromProperties(props);
+    MetricsConfig metricsConfig =
+        table != null ? MetricsConfig.forTable(table) : MetricsConfig.fromProperties(props);


this should be forPositionDelete

stevenzwu · 2022-12-13T02:53:35Z

flink/v1.16/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkAppenderFactory.java

  public FileAppender<RowData> newAppender(OutputFile outputFile, FileFormat format) {
-    MetricsConfig metricsConfig = MetricsConfig.fromProperties(props);
+    MetricsConfig metricsConfig =
+        table != null ? MetricsConfig.forTable(table) : MetricsConfig.fromProperties(props);


MetricsConfig.fromProperties is deprecated. I am wondering if we should just ensure table is never null? That would require changing the current 2 constructors (technically breaking) without adding a new constructor. but this class should be an @Internal class. what do you think? @pvary @hililiwei @chenjunjiedada

stevenzwu · 2022-12-14T16:33:48Z

thanks @chenjunjiedada for the contribution

github-actions bot added build core flink labels Nov 30, 2022

chenjunjiedada commented Nov 30, 2022

View reviewed changes

stevenzwu reviewed Nov 30, 2022

View reviewed changes

hililiwei approved these changes Dec 1, 2022

View reviewed changes

chenjunjiedada added 3 commits December 2, 2022 14:54

Flink: use correct metric config for position deletes

cbb3fb9

refactor

205638d

rebase

5d63141

chenjunjiedada force-pushed the flink-metrics-config branch from 904b90f to 5d63141 Compare December 2, 2022 07:05

refactor according to steven's comments

c44d036

chenjunjiedada commented Dec 6, 2022

View reviewed changes

tmp

d75c3d4

stevenzwu reviewed Dec 13, 2022

View reviewed changes

chenjunjiedada added 5 commits December 13, 2022 14:33

tmp

4b885c3

refactor ctor

0a9337f

fix for flink 1.14

a762972

fix flink 1.15

4de6115

add check

f43000b

stevenzwu approved these changes Dec 14, 2022

View reviewed changes

stevenzwu merged commit b735401 into apache:master Dec 14, 2022

zhongyujiang mentioned this pull request Jan 30, 2023

position delete manifest lower_bounds/upper_bounds not correct #6694

Closed

singhpk234 mentioned this pull request Apr 5, 2024

In case of Positional Deletes "file_path" in lowerbound & upperbound do not have full file_path Buffer #10064

Closed

zhongyujiang pushed a commit to zhongyujiang/iceberg that referenced this pull request Apr 16, 2025

Flink: use correct metric config for position deletes apache#6313

3b27ecc

Flink: use correct metric config for position deletes #6313

Flink: use correct metric config for position deletes #6313

Uh oh!

Conversation

chenjunjiedada commented Nov 30, 2022

Uh oh!

Choose a reason for hiding this comment

Uh oh!

chenjunjiedada commented Nov 30, 2022

Uh oh!

stevenzwu Nov 30, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

stevenzwu Dec 6, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

stevenzwu Dec 6, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hililiwei left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

stevenzwu Dec 13, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

stevenzwu commented Dec 14, 2022

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

stevenzwu Nov 30, 2022 •

edited

Loading

stevenzwu Dec 6, 2022 •

edited

Loading

stevenzwu Dec 6, 2022 •

edited

Loading

stevenzwu Dec 13, 2022 •

edited

Loading