You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "pvary (via GitHub)" <gi...@apache.org> on 2025/03/30 22:11:56 UTC
[PR] Spark: Update RewriteDataFilesSparkAction and RewritePositionDeleteFilesSparkAction to use the new APIs [iceberg]
pvary opened a new pull request, #12692:
URL: https://github.com/apache/iceberg/pull/12692
(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] Spark: Update RewriteDataFilesSparkAction and RewritePositionDeleteFilesSparkAction to use the new APIs [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #12692:
URL: https://github.com/apache/iceberg/pull/12692#issuecomment-2783741576
@danielcweeks, @manuzhang, @RussellSpitzer: I think this PR is ready for review now. Sorry for the force push. Instead of creating new files, I renamed the old ones so it is easier to review. This required me to force-push my changes. I will not do it again during 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] Spark: Update RewriteDataFilesSparkAction and RewritePositionDeleteFilesSparkAction to use the new APIs [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12692:
URL: https://github.com/apache/iceberg/pull/12692#discussion_r2067872681
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestRewriteDataFilesProcedure.java:
##########
@@ -638,7 +638,7 @@ public void testRewriteDataFilesWithInvalidInputs() {
+ "sort_order => 'c1 ASC NULLS FIRST')",
catalogName, tableIdent))
.isInstanceOf(IllegalArgumentException.class)
- .hasMessage("Must use only one rewriter type (bin-pack, sort, zorder)");
+ .hasMessageStartingWith("Rewriter type already set to ");
Review Comment:
Changed to `Cannot set rewrite mode, it has already been set to `. I this is a bit better, but ok with reverting your version if you don't agree.
--
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] Spark: Update RewriteDataFilesSparkAction and RewritePositionDeleteFilesSparkAction to use the new APIs [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12692:
URL: https://github.com/apache/iceberg/pull/12692#discussion_r2039897696
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java:
##########
@@ -101,9 +92,10 @@ public class RewriteDataFilesSparkAction
private boolean partialProgressEnabled;
private boolean removeDanglingDeletes;
private boolean useStartingSequenceNumber;
- private RewriteJobOrder rewriteJobOrder;
- private FileRewriter<FileScanTask, DataFile> rewriter = null;
private boolean caseSensitive;
+ private BinPackRewriteFilePlanner planner = null;
+ private FileRewriteRunner<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> runner = null;
+ private boolean shufflingPlanner = false;
Review Comment:
Why a separate variable here? Can't we just check "if Rewriter extends SortRewriter"
--
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] Spark: Update RewriteDataFilesSparkAction and RewritePositionDeleteFilesSparkAction to use the new APIs [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12692:
URL: https://github.com/apache/iceberg/pull/12692#discussion_r2041633459
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/SparkRewriteRunner.java:
##########
@@ -0,0 +1,79 @@
+/*
+ * 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.spark.actions;
+
+import java.util.Map;
+import java.util.Set;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.ContentScanTask;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.actions.FileRewriteRunner;
+import org.apache.iceberg.actions.RewriteDataFiles;
+import org.apache.iceberg.actions.RewriteFileGroup;
+import org.apache.iceberg.actions.RewriteGroupBase;
+import org.apache.iceberg.actions.RewritePositionDeleteFiles;
+import org.apache.iceberg.actions.RewritePositionDeletesGroup;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.spark.sql.SparkSession;
+
+/**
+ * Common parent for data and positional delete rewrite runners.
Review Comment:
The Action is using the 2 interfaces defined by the new API. Planner for grouping, Runner for executing the actual compaction.
This class is the base for the Spark based Runner implementations. Updated the javadoc
--
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] Spark: Update RewriteDataFilesSparkAction and RewritePositionDeleteFilesSparkAction to use the new APIs [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12692:
URL: https://github.com/apache/iceberg/pull/12692#discussion_r2039895111
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java:
##########
@@ -121,32 +113,36 @@ protected RewriteDataFilesSparkAction self() {
@Override
public RewriteDataFilesSparkAction binPack() {
Preconditions.checkArgument(
- rewriter == null, "Must use only one rewriter type (bin-pack, sort, zorder)");
- this.rewriter = new SparkBinPackDataRewriter(spark(), table);
+ runner == null, "Must use only one rewriter type (bin-pack, sort, zorder)");
Review Comment:
Follow up, We need to clean up these messages. They probably should say, "Rewriter type already set to %s" or something like that
--
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] Spark: Update RewriteDataFilesSparkAction and RewritePositionDeleteFilesSparkAction to use the new APIs [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12692:
URL: https://github.com/apache/iceberg/pull/12692#discussion_r2039899219
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java:
##########
@@ -196,69 +183,29 @@ public RewriteDataFiles.Result execute() {
return resultBuilder.build();
}
- StructLikeMap<List<List<FileScanTask>>> planFileGroups(long startingSnapshotId) {
- CloseableIterable<FileScanTask> fileScanTasks =
- table
- .newScan()
- .useSnapshot(startingSnapshotId)
- .caseSensitive(caseSensitive)
- .filter(filter)
- .ignoreResiduals()
- .planFiles();
+ private void init(long startingSnapshotId) {
- try {
- StructType partitionType = table.spec().partitionType();
- StructLikeMap<List<FileScanTask>> filesByPartition =
- groupByPartition(partitionType, fileScanTasks);
- return fileGroupsByPartition(filesByPartition);
- } finally {
- try {
- fileScanTasks.close();
- } catch (IOException io) {
- LOG.error("Cannot properly close file iterable while planning for rewrite", io);
- }
- }
- }
+ this.planner =
+ shufflingPlanner
Review Comment:
I noted above but I think we can just check if our rewrite extends a class that requires shuffles?
--
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] Spark: Update RewriteDataFilesSparkAction and RewritePositionDeleteFilesSparkAction to use the new APIs [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #12692:
URL: https://github.com/apache/iceberg/pull/12692#issuecomment-2842204671
Merged to main.
Thanks @manuzhang, @danielcweeks, @RussellSpitzer, @stevenzwu for the 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] Spark: Update RewriteDataFilesSparkAction and RewritePositionDeleteFilesSparkAction to use the new APIs [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12692:
URL: https://github.com/apache/iceberg/pull/12692#discussion_r2020489760
##########
spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java:
##########
@@ -1720,143 +1717,15 @@ public void testInvalidAPIUsage() {
assertThatThrownBy(() -> actions().rewriteDataFiles(table).binPack().sort())
.isInstanceOf(IllegalArgumentException.class)
- .hasMessage("Must use only one rewriter type (bin-pack, sort, zorder)");
+ .hasMessage("Must use only one runner type (bin-pack, sort, zorder)");
Review Comment:
I'm not sure about this change.
The internal naming has changed, but not sure we have to expose this to the users
--
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] Spark: Update RewriteDataFilesSparkAction and RewritePositionDeleteFilesSparkAction to use the new APIs [iceberg]
Posted by "manuzhang (via GitHub)" <gi...@apache.org>.
manuzhang commented on PR #12692:
URL: https://github.com/apache/iceberg/pull/12692#issuecomment-2833879096
I add this to 1.10.0, since we are targeting removing deprecated APIs in 1.10.0
--
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] Spark: Update RewriteDataFilesSparkAction and RewritePositionDeleteFilesSparkAction to use the new APIs [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #12692:
URL: https://github.com/apache/iceberg/pull/12692#discussion_r2067584705
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java:
##########
@@ -120,36 +111,39 @@ protected RewriteDataFilesSparkAction self() {
@Override
public RewriteDataFilesSparkAction binPack() {
- Preconditions.checkArgument(
- rewriter == null, "Must use only one rewriter type (bin-pack, sort, zorder)");
- this.rewriter = new SparkBinPackDataRewriter(spark(), table);
+ checkRunnerIsUnset();
+ this.runner = new SparkBinPackFileRewriteRunner(spark(), table);
return this;
}
@Override
public RewriteDataFilesSparkAction sort(SortOrder sortOrder) {
- Preconditions.checkArgument(
- rewriter == null, "Must use only one rewriter type (bin-pack, sort, zorder)");
- this.rewriter = new SparkSortDataRewriter(spark(), table, sortOrder);
+ checkRunnerIsUnset();
+ this.runner = new SparkSortFileRewriteRunner(spark(), table, sortOrder);
return this;
}
@Override
public RewriteDataFilesSparkAction sort() {
- Preconditions.checkArgument(
- rewriter == null, "Must use only one rewriter type (bin-pack, sort, zorder)");
- this.rewriter = new SparkSortDataRewriter(spark(), table);
+ checkRunnerIsUnset();
+ this.runner = new SparkSortFileRewriteRunner(spark(), table);
return this;
}
@Override
public RewriteDataFilesSparkAction zOrder(String... columnNames) {
- Preconditions.checkArgument(
- rewriter == null, "Must use only one rewriter type (bin-pack, sort, zorder)");
- this.rewriter = new SparkZOrderDataRewriter(spark(), table, Arrays.asList(columnNames));
+ checkRunnerIsUnset();
+ this.runner = new SparkZOrderFileRewriteRunner(spark(), table, Arrays.asList(columnNames));
return this;
}
+ private void checkRunnerIsUnset() {
+ Preconditions.checkArgument(
+ runner == null,
+ "Rewriter type already set to %s",
+ runner == null ? null : runner.description());
Review Comment:
when check fails, `runner` is ensured to be not null. do we need the null check?
--
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] Spark: Update RewriteDataFilesSparkAction and RewritePositionDeleteFilesSparkAction to use the new APIs [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12692:
URL: https://github.com/apache/iceberg/pull/12692#discussion_r2067872681
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestRewriteDataFilesProcedure.java:
##########
@@ -638,7 +638,7 @@ public void testRewriteDataFilesWithInvalidInputs() {
+ "sort_order => 'c1 ASC NULLS FIRST')",
catalogName, tableIdent))
.isInstanceOf(IllegalArgumentException.class)
- .hasMessage("Must use only one rewriter type (bin-pack, sort, zorder)");
+ .hasMessageStartingWith("Rewriter type already set to ");
Review Comment:
Changed to `Cannot set rewrite mode, it has already been set to `. I think this is a bit better, but ok with reverting to your version if you don't agree.
--
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] Spark: Update RewriteDataFilesSparkAction and RewritePositionDeleteFilesSparkAction to use the new APIs [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary merged PR #12692:
URL: https://github.com/apache/iceberg/pull/12692
--
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] Spark: Update RewriteDataFilesSparkAction and RewritePositionDeleteFilesSparkAction to use the new APIs [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12692:
URL: https://github.com/apache/iceberg/pull/12692#discussion_r2031502283
##########
spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java:
##########
@@ -1720,143 +1717,15 @@ public void testInvalidAPIUsage() {
assertThatThrownBy(() -> actions().rewriteDataFiles(table).binPack().sort())
.isInstanceOf(IllegalArgumentException.class)
- .hasMessage("Must use only one rewriter type (bin-pack, sort, zorder)");
+ .hasMessage("Must use only one runner type (bin-pack, sort, zorder)");
Review Comment:
I decided to keep this error message intact.
--
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] Spark: Update RewriteDataFilesSparkAction and RewritePositionDeleteFilesSparkAction to use the new APIs [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12692:
URL: https://github.com/apache/iceberg/pull/12692#discussion_r2041639954
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/SparkShufflingDataRewritePlanner.java:
##########
@@ -0,0 +1,81 @@
+/*
+ * 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.spark.actions;
+
+import java.util.Map;
+import java.util.Set;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.actions.BinPackRewriteFilePlanner;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.util.PropertyUtil;
+
+/**
+ * Extends the {@link BinPackRewriteFilePlanner} with the possibility to set the expected
Review Comment:
Again, this is for the Planner. There was a specific configuration for the Shuffling rewriters which is modifying the plans. So I had to extend the core planners with the new configuration.
Alternatively we can move this functionality to the generic planner. In this case we don't need a specific class here. In this case the we have this config available for every planner (might be useful if there are compaction changes on any rewrite), but we have to reintroduce the flag `shufflingPlanner` (maybe with another name) to decide which Runner implementation to use.
--
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] Spark: Update RewriteDataFilesSparkAction and RewritePositionDeleteFilesSparkAction to use the new APIs [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12692:
URL: https://github.com/apache/iceberg/pull/12692#discussion_r2064023207
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/SparkBinPackFileRewriteRunner.java:
##########
@@ -56,16 +58,18 @@ protected void doRewrite(String groupId, List<FileScanTask> group) {
.write()
.format("iceberg")
.option(SparkWriteOptions.REWRITTEN_FILE_SCAN_TASK_SET_ID, groupId)
- .option(SparkWriteOptions.TARGET_FILE_SIZE_BYTES, writeMaxFileSize())
- .option(SparkWriteOptions.DISTRIBUTION_MODE, distributionMode(group).modeName())
- .option(SparkWriteOptions.OUTPUT_SPEC_ID, outputSpecId())
+ .option(SparkWriteOptions.TARGET_FILE_SIZE_BYTES, group.maxOutputFileSize())
+ .option(
+ SparkWriteOptions.DISTRIBUTION_MODE,
+ distributionMode(group.fileScanTasks(), spec(group.outputSpecId())).modeName())
+ .option(SparkWriteOptions.OUTPUT_SPEC_ID, group.outputSpecId())
.mode("append")
.save(groupId);
}
// invoke a shuffle if the original spec does not match the output spec
- private DistributionMode distributionMode(List<FileScanTask> group) {
- boolean requiresRepartition = !group.get(0).spec().equals(outputSpec());
+ private DistributionMode distributionMode(List<FileScanTask> group, PartitionSpec outputSpec) {
Review Comment:
rather than adding more parameters and extracting more fields from "group" why not just pass group all the way through? Would simplify line 64 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] Spark: Update RewriteDataFilesSparkAction and RewritePositionDeleteFilesSparkAction to use the new APIs [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12692:
URL: https://github.com/apache/iceberg/pull/12692#discussion_r2063996698
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/SparkShufflingDataRewritePlanner.java:
##########
@@ -0,0 +1,81 @@
+/*
+ * 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.spark.actions;
+
+import java.util.Map;
+import java.util.Set;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.actions.BinPackRewriteFilePlanner;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.util.PropertyUtil;
+
+/**
+ * Extends the {@link BinPackRewriteFilePlanner} with the possibility to set the expected
Review Comment:
I was just talking about the Java Doc. I have no problem with the code organization but the key detail about this planner is that it produces plans for Shuffling Rewrites, not that it has an additional option.
--
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] Spark: Update RewriteDataFilesSparkAction and RewritePositionDeleteFilesSparkAction to use the new APIs [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12692:
URL: https://github.com/apache/iceberg/pull/12692#discussion_r2067879419
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/SparkBinPackFileRewriteRunner.java:
##########
@@ -56,16 +58,18 @@ protected void doRewrite(String groupId, List<FileScanTask> group) {
.write()
.format("iceberg")
.option(SparkWriteOptions.REWRITTEN_FILE_SCAN_TASK_SET_ID, groupId)
- .option(SparkWriteOptions.TARGET_FILE_SIZE_BYTES, writeMaxFileSize())
- .option(SparkWriteOptions.DISTRIBUTION_MODE, distributionMode(group).modeName())
- .option(SparkWriteOptions.OUTPUT_SPEC_ID, outputSpecId())
+ .option(SparkWriteOptions.TARGET_FILE_SIZE_BYTES, group.maxOutputFileSize())
+ .option(
+ SparkWriteOptions.DISTRIBUTION_MODE,
+ distributionMode(group.fileScanTasks(), spec(group.outputSpecId())).modeName())
+ .option(SparkWriteOptions.OUTPUT_SPEC_ID, group.outputSpecId())
.mode("append")
.save(groupId);
}
// invoke a shuffle if the original spec does not match the output spec
- private DistributionMode distributionMode(List<FileScanTask> group) {
- boolean requiresRepartition = !group.get(0).spec().equals(outputSpec());
+ private DistributionMode distributionMode(List<FileScanTask> group, PartitionSpec outputSpec) {
Review Comment:
Good idea.
Done
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java:
##########
@@ -120,36 +111,39 @@ protected RewriteDataFilesSparkAction self() {
@Override
public RewriteDataFilesSparkAction binPack() {
- Preconditions.checkArgument(
- rewriter == null, "Must use only one rewriter type (bin-pack, sort, zorder)");
- this.rewriter = new SparkBinPackDataRewriter(spark(), table);
+ checkRunnerIsUnset();
+ this.runner = new SparkBinPackFileRewriteRunner(spark(), table);
return this;
}
@Override
public RewriteDataFilesSparkAction sort(SortOrder sortOrder) {
- Preconditions.checkArgument(
- rewriter == null, "Must use only one rewriter type (bin-pack, sort, zorder)");
- this.rewriter = new SparkSortDataRewriter(spark(), table, sortOrder);
+ checkRunnerIsUnset();
+ this.runner = new SparkSortFileRewriteRunner(spark(), table, sortOrder);
return this;
}
@Override
public RewriteDataFilesSparkAction sort() {
- Preconditions.checkArgument(
- rewriter == null, "Must use only one rewriter type (bin-pack, sort, zorder)");
- this.rewriter = new SparkSortDataRewriter(spark(), table);
+ checkRunnerIsUnset();
+ this.runner = new SparkSortFileRewriteRunner(spark(), table);
return this;
}
@Override
public RewriteDataFilesSparkAction zOrder(String... columnNames) {
- Preconditions.checkArgument(
- rewriter == null, "Must use only one rewriter type (bin-pack, sort, zorder)");
- this.rewriter = new SparkZOrderDataRewriter(spark(), table, Arrays.asList(columnNames));
+ checkRunnerIsUnset();
+ this.runner = new SparkZOrderFileRewriteRunner(spark(), table, Arrays.asList(columnNames));
return this;
}
+ private void checkRunnerIsUnset() {
Review Comment:
Done
--
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] Spark: Update RewriteDataFilesSparkAction and RewritePositionDeleteFilesSparkAction to use the new APIs [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12692:
URL: https://github.com/apache/iceberg/pull/12692#discussion_r2067879186
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/SparkShufflingDataRewritePlanner.java:
##########
@@ -0,0 +1,81 @@
+/*
+ * 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.spark.actions;
+
+import java.util.Map;
+import java.util.Set;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.actions.BinPackRewriteFilePlanner;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.util.PropertyUtil;
+
+/**
+ * Extends the {@link BinPackRewriteFilePlanner} with the possibility to set the expected
Review Comment:
Thanks for all the suggestions.
Updated the javadoc based on your input.
--
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] Spark: Update RewriteDataFilesSparkAction and RewritePositionDeleteFilesSparkAction to use the new APIs [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12692:
URL: https://github.com/apache/iceberg/pull/12692#discussion_r2040392585
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/SparkShufflingDataRewritePlanner.java:
##########
@@ -0,0 +1,81 @@
+/*
+ * 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.spark.actions;
+
+import java.util.Map;
+import java.util.Set;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.actions.BinPackRewriteFilePlanner;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.util.PropertyUtil;
+
+/**
+ * Extends the {@link BinPackRewriteFilePlanner} with the possibility to set the expected
Review Comment:
Needs more description here. Extends the BinPack rewriter for Rewriters that induce a distributed shuffle to reorganize data. (or something like 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] Spark: Update RewriteDataFilesSparkAction and RewritePositionDeleteFilesSparkAction to use the new APIs [iceberg]
Posted by "danielcweeks (via GitHub)" <gi...@apache.org>.
danielcweeks commented on code in PR #12692:
URL: https://github.com/apache/iceberg/pull/12692#discussion_r2021463331
##########
spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java:
##########
@@ -1720,143 +1717,15 @@ public void testInvalidAPIUsage() {
assertThatThrownBy(() -> actions().rewriteDataFiles(table).binPack().sort())
.isInstanceOf(IllegalArgumentException.class)
- .hasMessage("Must use only one rewriter type (bin-pack, sort, zorder)");
+ .hasMessage("Must use only one runner type (bin-pack, sort, zorder)");
Review Comment:
I wouldn't go with the "runner" term. That's not really a concept in Spark, so I think this just exposes the internals in a confusing way.
--
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] Spark: Update RewriteDataFilesSparkAction and RewritePositionDeleteFilesSparkAction to use the new APIs [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12692:
URL: https://github.com/apache/iceberg/pull/12692#discussion_r2064074301
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestRewriteDataFilesProcedure.java:
##########
@@ -638,7 +638,7 @@ public void testRewriteDataFilesWithInvalidInputs() {
+ "sort_order => 'c1 ASC NULLS FIRST')",
catalogName, tableIdent))
.isInstanceOf(IllegalArgumentException.class)
- .hasMessage("Must use only one rewriter type (bin-pack, sort, zorder)");
+ .hasMessageStartingWith("Rewriter type already set to ");
Review Comment:
I know this was not correct before but can we switch the error message here to "Cannot set rewriter, it has already been set to "
--
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] Spark: Update RewriteDataFilesSparkAction and RewritePositionDeleteFilesSparkAction to use the new APIs [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #12692:
URL: https://github.com/apache/iceberg/pull/12692#discussion_r2067584705
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java:
##########
@@ -120,36 +111,39 @@ protected RewriteDataFilesSparkAction self() {
@Override
public RewriteDataFilesSparkAction binPack() {
- Preconditions.checkArgument(
- rewriter == null, "Must use only one rewriter type (bin-pack, sort, zorder)");
- this.rewriter = new SparkBinPackDataRewriter(spark(), table);
+ checkRunnerIsUnset();
+ this.runner = new SparkBinPackFileRewriteRunner(spark(), table);
return this;
}
@Override
public RewriteDataFilesSparkAction sort(SortOrder sortOrder) {
- Preconditions.checkArgument(
- rewriter == null, "Must use only one rewriter type (bin-pack, sort, zorder)");
- this.rewriter = new SparkSortDataRewriter(spark(), table, sortOrder);
+ checkRunnerIsUnset();
+ this.runner = new SparkSortFileRewriteRunner(spark(), table, sortOrder);
return this;
}
@Override
public RewriteDataFilesSparkAction sort() {
- Preconditions.checkArgument(
- rewriter == null, "Must use only one rewriter type (bin-pack, sort, zorder)");
- this.rewriter = new SparkSortDataRewriter(spark(), table);
+ checkRunnerIsUnset();
+ this.runner = new SparkSortFileRewriteRunner(spark(), table);
return this;
}
@Override
public RewriteDataFilesSparkAction zOrder(String... columnNames) {
- Preconditions.checkArgument(
- rewriter == null, "Must use only one rewriter type (bin-pack, sort, zorder)");
- this.rewriter = new SparkZOrderDataRewriter(spark(), table, Arrays.asList(columnNames));
+ checkRunnerIsUnset();
+ this.runner = new SparkZOrderFileRewriteRunner(spark(), table, Arrays.asList(columnNames));
return this;
}
+ private void checkRunnerIsUnset() {
+ Preconditions.checkArgument(
+ runner == null,
+ "Rewriter type already set to %s",
+ runner == null ? null : runner.description());
Review Comment:
when check fails, `runner` is ensured to be not null. do we need the null check?
--
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] Spark: Update RewriteDataFilesSparkAction and RewritePositionDeleteFilesSparkAction to use the new APIs [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #12692:
URL: https://github.com/apache/iceberg/pull/12692#discussion_r2067626037
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/SparkShufflingDataRewritePlanner.java:
##########
@@ -0,0 +1,81 @@
+/*
+ * 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.spark.actions;
+
+import java.util.Map;
+import java.util.Set;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.actions.BinPackRewriteFilePlanner;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.util.PropertyUtil;
+
+/**
+ * Extends the {@link BinPackRewriteFilePlanner} with the possibility to set the expected
Review Comment:
I thought this class adds an additional option of `COMPRESSION_FACTOR` used when calculating `expectedOutputFiles`
--
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] Spark: Update RewriteDataFilesSparkAction and RewritePositionDeleteFilesSparkAction to use the new APIs [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12692:
URL: https://github.com/apache/iceberg/pull/12692#discussion_r2040391797
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/SparkRewriteRunner.java:
##########
@@ -0,0 +1,79 @@
+/*
+ * 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.spark.actions;
+
+import java.util.Map;
+import java.util.Set;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.ContentScanTask;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.actions.FileRewriteRunner;
+import org.apache.iceberg.actions.RewriteDataFiles;
+import org.apache.iceberg.actions.RewriteFileGroup;
+import org.apache.iceberg.actions.RewriteGroupBase;
+import org.apache.iceberg.actions.RewritePositionDeleteFiles;
+import org.apache.iceberg.actions.RewritePositionDeletesGroup;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.spark.sql.SparkSession;
+
+/**
+ * Common parent for data and positional delete rewrite runners.
Review Comment:
Unclear description here, I think this is mean to essentially replace the meet of the Action itself? Like it's a container for planner + rewriter?
--
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] Spark: Update RewriteDataFilesSparkAction and RewritePositionDeleteFilesSparkAction to use the new APIs [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #12692:
URL: https://github.com/apache/iceberg/pull/12692#discussion_r2067626037
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/SparkShufflingDataRewritePlanner.java:
##########
@@ -0,0 +1,81 @@
+/*
+ * 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.spark.actions;
+
+import java.util.Map;
+import java.util.Set;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.actions.BinPackRewriteFilePlanner;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.util.PropertyUtil;
+
+/**
+ * Extends the {@link BinPackRewriteFilePlanner} with the possibility to set the expected
Review Comment:
I thought this class adds an additional option of `COMPRESSION_FACTOR` used when calculating `expectedOutputFiles` (line 60). I guess shuffle/sort improves compression ratio and reduce file size.
--
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] Spark: Update RewriteDataFilesSparkAction and RewritePositionDeleteFilesSparkAction to use the new APIs [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #12692:
URL: https://github.com/apache/iceberg/pull/12692#discussion_r2067584229
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java:
##########
@@ -120,36 +111,39 @@ protected RewriteDataFilesSparkAction self() {
@Override
public RewriteDataFilesSparkAction binPack() {
- Preconditions.checkArgument(
- rewriter == null, "Must use only one rewriter type (bin-pack, sort, zorder)");
- this.rewriter = new SparkBinPackDataRewriter(spark(), table);
+ checkRunnerIsUnset();
+ this.runner = new SparkBinPackFileRewriteRunner(spark(), table);
return this;
}
@Override
public RewriteDataFilesSparkAction sort(SortOrder sortOrder) {
- Preconditions.checkArgument(
- rewriter == null, "Must use only one rewriter type (bin-pack, sort, zorder)");
- this.rewriter = new SparkSortDataRewriter(spark(), table, sortOrder);
+ checkRunnerIsUnset();
+ this.runner = new SparkSortFileRewriteRunner(spark(), table, sortOrder);
return this;
}
@Override
public RewriteDataFilesSparkAction sort() {
- Preconditions.checkArgument(
- rewriter == null, "Must use only one rewriter type (bin-pack, sort, zorder)");
- this.rewriter = new SparkSortDataRewriter(spark(), table);
+ checkRunnerIsUnset();
+ this.runner = new SparkSortFileRewriteRunner(spark(), table);
return this;
}
@Override
public RewriteDataFilesSparkAction zOrder(String... columnNames) {
- Preconditions.checkArgument(
- rewriter == null, "Must use only one rewriter type (bin-pack, sort, zorder)");
- this.rewriter = new SparkZOrderDataRewriter(spark(), table, Arrays.asList(columnNames));
+ checkRunnerIsUnset();
+ this.runner = new SparkZOrderFileRewriteRunner(spark(), table, Arrays.asList(columnNames));
return this;
}
+ private void checkRunnerIsUnset() {
Review Comment:
nit: `ensureRunnerNotSet` is probably a little more clear
--
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] Spark: Update RewriteDataFilesSparkAction and RewritePositionDeleteFilesSparkAction to use the new APIs [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12692:
URL: https://github.com/apache/iceberg/pull/12692#discussion_r2040331964
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java:
##########
@@ -121,32 +113,36 @@ protected RewriteDataFilesSparkAction self() {
@Override
public RewriteDataFilesSparkAction binPack() {
Preconditions.checkArgument(
- rewriter == null, "Must use only one rewriter type (bin-pack, sort, zorder)");
- this.rewriter = new SparkBinPackDataRewriter(spark(), table);
+ runner == null, "Must use only one rewriter type (bin-pack, sort, zorder)");
Review Comment:
This is a second request to fix the error message, and it is not too complicated. So changed.
The solution is a bit lame:
```
Preconditions.checkArgument(
runner == null,
"Rewriter type already set to %s",
runner == null ? null : runner.description());
```
We need a second null check for the error message, or a null check around it 😢
Decided to hide this ugliness in a method.
If you have better ideas, feel free to comment
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java:
##########
@@ -101,9 +92,10 @@ public class RewriteDataFilesSparkAction
private boolean partialProgressEnabled;
private boolean removeDanglingDeletes;
private boolean useStartingSequenceNumber;
- private RewriteJobOrder rewriteJobOrder;
- private FileRewriter<FileScanTask, DataFile> rewriter = null;
private boolean caseSensitive;
+ private BinPackRewriteFilePlanner planner = null;
+ private FileRewriteRunner<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> runner = null;
+ private boolean shufflingPlanner = false;
Review Comment:
Good idea.
Done
--
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] Spark: Update RewriteDataFilesSparkAction and RewritePositionDeleteFilesSparkAction to use the new APIs [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12692:
URL: https://github.com/apache/iceberg/pull/12692#discussion_r2040332211
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java:
##########
@@ -196,69 +183,29 @@ public RewriteDataFiles.Result execute() {
return resultBuilder.build();
}
- StructLikeMap<List<List<FileScanTask>>> planFileGroups(long startingSnapshotId) {
- CloseableIterable<FileScanTask> fileScanTasks =
- table
- .newScan()
- .useSnapshot(startingSnapshotId)
- .caseSensitive(caseSensitive)
- .filter(filter)
- .ignoreResiduals()
- .planFiles();
+ private void init(long startingSnapshotId) {
- try {
- StructType partitionType = table.spec().partitionType();
- StructLikeMap<List<FileScanTask>> filesByPartition =
- groupByPartition(partitionType, fileScanTasks);
- return fileGroupsByPartition(filesByPartition);
- } finally {
- try {
- fileScanTasks.close();
- } catch (IOException io) {
- LOG.error("Cannot properly close file iterable while planning for rewrite", io);
- }
- }
- }
+ this.planner =
+ shufflingPlanner
Review Comment:
Done
--
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] Spark: Update RewriteDataFilesSparkAction and RewritePositionDeleteFilesSparkAction to use the new APIs [iceberg]
Posted by "manuzhang (via GitHub)" <gi...@apache.org>.
manuzhang commented on code in PR #12692:
URL: https://github.com/apache/iceberg/pull/12692#discussion_r2024844226
##########
spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java:
##########
@@ -1720,143 +1717,15 @@ public void testInvalidAPIUsage() {
assertThatThrownBy(() -> actions().rewriteDataFiles(table).binPack().sort())
.isInstanceOf(IllegalArgumentException.class)
- .hasMessage("Must use only one rewriter type (bin-pack, sort, zorder)");
+ .hasMessage("Must use only one runner type (bin-pack, sort, zorder)");
Review Comment:
We use "strategy" in the [user doc](https://iceberg.apache.org/docs/nightly/spark-procedures/#usage_9).
--
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