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/10 15:52:02 UTC
[PR] Core: FileRewritePlanner implementation [iceberg]
pvary opened a new pull request, #12493:
URL: https://github.com/apache/iceberg/pull/12493
This PR creates the core implementations for the Planners defined by #12306.
The PR deprecates the old `*Rewriter` implementations. Adds test for the Planners.
This is part of the #11513 refactor
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2016143274
##########
core/src/test/java/org/apache/iceberg/actions/TestBinPackRewriteFileGroupPlanner.java:
##########
@@ -0,0 +1,468 @@
+/*
+ * 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.actions;
+
+import static org.apache.iceberg.actions.BinPackRewriteFileGroupPlanner.MAX_FILE_SIZE_DEFAULT_RATIO;
+import static org.apache.iceberg.actions.RewriteDataFiles.REWRITE_JOB_ORDER;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.File;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.stream.Collectors;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.MockFileScanTask;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.TestBase;
+import org.apache.iceberg.TestTables;
+import org.apache.iceberg.actions.RewriteDataFiles.FileGroupInfo;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class TestBinPackRewriteFileGroupPlanner {
+ private static final Map<String, String> REWRITE_ALL =
+ ImmutableMap.of(BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true");
+
+ private static final DataFile FILE_1 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_2 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_3 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_4 = newDataFile("data_bucket=1", 11);
+ private static final DataFile FILE_5 = newDataFile("data_bucket=1", 11);
+ private static final DataFile FILE_6 = newDataFile("data_bucket=2", 50);
+
+ private static final Map<RewriteJobOrder, List<StructLike>> EXPECTED =
+ ImmutableMap.of(
+ RewriteJobOrder.FILES_DESC,
+ ImmutableList.of(FILE_1.partition(), FILE_4.partition(), FILE_6.partition()),
+ RewriteJobOrder.FILES_ASC,
+ ImmutableList.of(FILE_6.partition(), FILE_4.partition(), FILE_1.partition()),
+ RewriteJobOrder.BYTES_DESC,
+ ImmutableList.of(FILE_6.partition(), FILE_1.partition(), FILE_4.partition()),
+ RewriteJobOrder.BYTES_ASC,
+ ImmutableList.of(FILE_4.partition(), FILE_1.partition(), FILE_6.partition()));
+
+ @TempDir private File tableDir = null;
+ private TestTables.TestTable table = null;
+
+ @BeforeEach
+ public void setupTable() throws Exception {
+ this.table = TestTables.create(tableDir, "test", TestBase.SCHEMA, TestBase.SPEC, 3);
+ }
+
+ @AfterEach
+ public void cleanupTables() {
+ TestTables.clearTables();
+ }
+
+ @Test
+ void testPartitionedTable() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(ImmutableMap.of(BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups).hasSize(3);
+ assertThat(groups.stream().mapToInt(RewriteGroupBase::inputFileNum).sum()).isEqualTo(6);
+ }
+
+ @Test
+ void testUnpartitionedTable() {
+ table.updateSpec().removeField("data_bucket").commit();
+ table.refresh();
+ table
+ .newAppend()
+ .appendFile(newDataFile("", 10))
+ .appendFile(newDataFile("", 20))
+ .appendFile(newDataFile("", 30))
+ .commit();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.MIN_INPUT_FILES,
+ "1",
+ BinPackRewriteFileGroupPlanner.MIN_FILE_SIZE_BYTES,
+ "30"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.totalGroupCount()).isEqualTo(1);
+ assertThat(plan.groups().iterator().next().inputFileNum()).isEqualTo(2);
+ }
+
+ @Test
+ void testEmptyTable() {
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(REWRITE_ALL);
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(table.currentSnapshot()).as("Table must be empty").isNull();
+ assertThat(plan.totalGroupCount()).isZero();
+ }
+
+ @ParameterizedTest
+ @EnumSource(
+ value = RewriteJobOrder.class,
+ names = {"FILES_DESC", "FILES_ASC", "BYTES_DESC", "BYTES_ASC"})
+ void testJobOrder(RewriteJobOrder order) {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true", REWRITE_JOB_ORDER, order.name()));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups.stream().map(group -> group.info().partition()).collect(Collectors.toList()))
+ .isEqualTo(EXPECTED.get(order));
+ assertThat(plan.totalGroupCount()).isEqualTo(3);
+ EXPECTED.get(order).forEach(s -> assertThat(plan.groupsInPartition(s)).isEqualTo(1));
+ }
+
+ @Test
+ void testMaxGroupSize() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.MAX_FILE_GROUP_SIZE_BYTES,
+ "10"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.totalGroupCount()).isEqualTo(6);
+ assertThat(plan.groupsInPartition(FILE_1.partition())).isEqualTo(3);
+ assertThat(plan.groupsInPartition(FILE_4.partition())).isEqualTo(2);
+ assertThat(plan.groupsInPartition(FILE_6.partition())).isEqualTo(1);
+ }
+
+ @Test
+ void testFilter() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner =
+ new BinPackRewriteFileGroupPlanner(
+ table,
+ Expressions.or(
+ Expressions.equal(Expressions.bucket("data", 16), 0),
+ Expressions.equal(Expressions.bucket("data", 16), 2)));
+ planner.init(REWRITE_ALL);
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(plan.totalGroupCount()).isEqualTo(2);
+ assertThat(groups).hasSize(2);
+ assertThat(groups.stream().mapToLong(RewriteFileGroup::inputFileNum).sum()).isEqualTo(4);
+ }
+
+ @Test
+ void testMaxOutputFileSize() {
+ addFiles();
+ int targetFileSize = 10;
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.TARGET_FILE_SIZE_BYTES,
+ String.valueOf(targetFileSize)));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.groups().iterator().next().maxOutputFileSize())
+ .isGreaterThan(targetFileSize)
+ .isLessThan((long) (targetFileSize * MAX_FILE_SIZE_DEFAULT_RATIO));
+ }
+
+ @Test
+ void testExpectedOutputFiles() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.TARGET_FILE_SIZE_BYTES,
+ "21"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups).hasSize(3);
+ for (RewriteFileGroup group : groups) {
+ if (FILE_1.partition().equals(group.info().partition())) {
+ assertThat(group.expectedOutputFiles()).isEqualTo(2);
+ } else if (FILE_4.partition().equals(group.info().partition())) {
+ assertThat(group.expectedOutputFiles()).isEqualTo(1);
+ } else if (FILE_6.partition().equals(group.info().partition())) {
+ assertThat(group.expectedOutputFiles()).isEqualTo(3);
+ } else {
+ throw new IllegalStateException("Unexpected partition: " + group.info().partition());
+ }
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testOutputSpec(boolean specific) {
+ addFiles();
+ int oldSpecId = table.spec().specId();
+ table.updateSpec().removeField("data_bucket").commit();
+ table.newAppend().appendFile(newDataFile("", 10)).commit();
+ table.refresh();
+ int newSpecId = table.spec().specId();
+
+ Map<String, String> options = Maps.newHashMap(REWRITE_ALL);
+ if (specific) {
+ options.put(RewriteDataFiles.OUTPUT_SPEC_ID, String.valueOf(oldSpecId));
+ }
+
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(options);
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.groups().iterator().next().outputSpecId())
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2016084243
##########
core/src/main/java/org/apache/iceberg/actions/BinPackRewriteFileGroupPlanner.java:
##########
@@ -0,0 +1,310 @@
+/*
+ * 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.actions;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.actions.RewriteDataFiles.FileGroupInfo;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.ContentFileUtil;
+import org.apache.iceberg.util.PropertyUtil;
+import org.apache.iceberg.util.StructLikeMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Groups specified data files in the {@link Table} into {@link RewriteFileGroup}s. The files are
+ * grouped by partitions based on their size using fix sized bins. Extends {@link
+ * SizeBasedFileRewritePlanner} with delete file number and delete ratio thresholds and job {@link
+ * RewriteDataFiles#REWRITE_JOB_ORDER} handling.
+ */
+public class BinPackRewriteFileGroupPlanner
Review Comment:
Renamed
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2015142847
##########
core/src/test/java/org/apache/iceberg/actions/TestBinPackRewriteFileGroupPlanner.java:
##########
@@ -0,0 +1,468 @@
+/*
+ * 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.actions;
+
+import static org.apache.iceberg.actions.BinPackRewriteFileGroupPlanner.MAX_FILE_SIZE_DEFAULT_RATIO;
+import static org.apache.iceberg.actions.RewriteDataFiles.REWRITE_JOB_ORDER;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.File;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.stream.Collectors;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.MockFileScanTask;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.TestBase;
+import org.apache.iceberg.TestTables;
+import org.apache.iceberg.actions.RewriteDataFiles.FileGroupInfo;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class TestBinPackRewriteFileGroupPlanner {
+ private static final Map<String, String> REWRITE_ALL =
+ ImmutableMap.of(BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true");
+
+ private static final DataFile FILE_1 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_2 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_3 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_4 = newDataFile("data_bucket=1", 11);
+ private static final DataFile FILE_5 = newDataFile("data_bucket=1", 11);
+ private static final DataFile FILE_6 = newDataFile("data_bucket=2", 50);
+
+ private static final Map<RewriteJobOrder, List<StructLike>> EXPECTED =
+ ImmutableMap.of(
+ RewriteJobOrder.FILES_DESC,
+ ImmutableList.of(FILE_1.partition(), FILE_4.partition(), FILE_6.partition()),
+ RewriteJobOrder.FILES_ASC,
+ ImmutableList.of(FILE_6.partition(), FILE_4.partition(), FILE_1.partition()),
+ RewriteJobOrder.BYTES_DESC,
+ ImmutableList.of(FILE_6.partition(), FILE_1.partition(), FILE_4.partition()),
+ RewriteJobOrder.BYTES_ASC,
+ ImmutableList.of(FILE_4.partition(), FILE_1.partition(), FILE_6.partition()));
+
+ @TempDir private File tableDir = null;
+ private TestTables.TestTable table = null;
+
+ @BeforeEach
+ public void setupTable() throws Exception {
+ this.table = TestTables.create(tableDir, "test", TestBase.SCHEMA, TestBase.SPEC, 3);
+ }
+
+ @AfterEach
+ public void cleanupTables() {
+ TestTables.clearTables();
+ }
+
+ @Test
+ void testPartitionedTable() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(ImmutableMap.of(BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups).hasSize(3);
+ assertThat(groups.stream().mapToInt(RewriteGroupBase::inputFileNum).sum()).isEqualTo(6);
+ }
+
+ @Test
+ void testUnpartitionedTable() {
+ table.updateSpec().removeField("data_bucket").commit();
+ table.refresh();
+ table
+ .newAppend()
+ .appendFile(newDataFile("", 10))
+ .appendFile(newDataFile("", 20))
+ .appendFile(newDataFile("", 30))
+ .commit();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.MIN_INPUT_FILES,
+ "1",
+ BinPackRewriteFileGroupPlanner.MIN_FILE_SIZE_BYTES,
+ "30"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.totalGroupCount()).isEqualTo(1);
+ assertThat(plan.groups().iterator().next().inputFileNum()).isEqualTo(2);
+ }
+
+ @Test
+ void testEmptyTable() {
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(REWRITE_ALL);
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(table.currentSnapshot()).as("Table must be empty").isNull();
+ assertThat(plan.totalGroupCount()).isZero();
+ }
+
+ @ParameterizedTest
+ @EnumSource(
+ value = RewriteJobOrder.class,
+ names = {"FILES_DESC", "FILES_ASC", "BYTES_DESC", "BYTES_ASC"})
+ void testJobOrder(RewriteJobOrder order) {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true", REWRITE_JOB_ORDER, order.name()));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups.stream().map(group -> group.info().partition()).collect(Collectors.toList()))
+ .isEqualTo(EXPECTED.get(order));
+ assertThat(plan.totalGroupCount()).isEqualTo(3);
+ EXPECTED.get(order).forEach(s -> assertThat(plan.groupsInPartition(s)).isEqualTo(1));
+ }
+
+ @Test
+ void testMaxGroupSize() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.MAX_FILE_GROUP_SIZE_BYTES,
+ "10"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.totalGroupCount()).isEqualTo(6);
+ assertThat(plan.groupsInPartition(FILE_1.partition())).isEqualTo(3);
+ assertThat(plan.groupsInPartition(FILE_4.partition())).isEqualTo(2);
+ assertThat(plan.groupsInPartition(FILE_6.partition())).isEqualTo(1);
+ }
+
+ @Test
+ void testFilter() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner =
+ new BinPackRewriteFileGroupPlanner(
+ table,
+ Expressions.or(
+ Expressions.equal(Expressions.bucket("data", 16), 0),
+ Expressions.equal(Expressions.bucket("data", 16), 2)));
+ planner.init(REWRITE_ALL);
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(plan.totalGroupCount()).isEqualTo(2);
Review Comment:
maybe add a little comment on which partitions would the filter match
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#issuecomment-2762698906
I haven't had time to check the rest of this, but I trust @stevenzwu 's review so please go on without me.
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2015157484
##########
core/src/test/java/org/apache/iceberg/actions/TestBinPackRewriteFileGroupPlanner.java:
##########
@@ -0,0 +1,468 @@
+/*
+ * 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.actions;
+
+import static org.apache.iceberg.actions.BinPackRewriteFileGroupPlanner.MAX_FILE_SIZE_DEFAULT_RATIO;
+import static org.apache.iceberg.actions.RewriteDataFiles.REWRITE_JOB_ORDER;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.File;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.stream.Collectors;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.MockFileScanTask;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.TestBase;
+import org.apache.iceberg.TestTables;
+import org.apache.iceberg.actions.RewriteDataFiles.FileGroupInfo;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class TestBinPackRewriteFileGroupPlanner {
+ private static final Map<String, String> REWRITE_ALL =
+ ImmutableMap.of(BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true");
+
+ private static final DataFile FILE_1 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_2 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_3 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_4 = newDataFile("data_bucket=1", 11);
+ private static final DataFile FILE_5 = newDataFile("data_bucket=1", 11);
+ private static final DataFile FILE_6 = newDataFile("data_bucket=2", 50);
+
+ private static final Map<RewriteJobOrder, List<StructLike>> EXPECTED =
+ ImmutableMap.of(
+ RewriteJobOrder.FILES_DESC,
+ ImmutableList.of(FILE_1.partition(), FILE_4.partition(), FILE_6.partition()),
+ RewriteJobOrder.FILES_ASC,
+ ImmutableList.of(FILE_6.partition(), FILE_4.partition(), FILE_1.partition()),
+ RewriteJobOrder.BYTES_DESC,
+ ImmutableList.of(FILE_6.partition(), FILE_1.partition(), FILE_4.partition()),
+ RewriteJobOrder.BYTES_ASC,
+ ImmutableList.of(FILE_4.partition(), FILE_1.partition(), FILE_6.partition()));
+
+ @TempDir private File tableDir = null;
+ private TestTables.TestTable table = null;
+
+ @BeforeEach
+ public void setupTable() throws Exception {
+ this.table = TestTables.create(tableDir, "test", TestBase.SCHEMA, TestBase.SPEC, 3);
+ }
+
+ @AfterEach
+ public void cleanupTables() {
+ TestTables.clearTables();
+ }
+
+ @Test
+ void testPartitionedTable() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(ImmutableMap.of(BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups).hasSize(3);
+ assertThat(groups.stream().mapToInt(RewriteGroupBase::inputFileNum).sum()).isEqualTo(6);
+ }
+
+ @Test
+ void testUnpartitionedTable() {
+ table.updateSpec().removeField("data_bucket").commit();
+ table.refresh();
+ table
+ .newAppend()
+ .appendFile(newDataFile("", 10))
+ .appendFile(newDataFile("", 20))
+ .appendFile(newDataFile("", 30))
+ .commit();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.MIN_INPUT_FILES,
+ "1",
+ BinPackRewriteFileGroupPlanner.MIN_FILE_SIZE_BYTES,
+ "30"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.totalGroupCount()).isEqualTo(1);
+ assertThat(plan.groups().iterator().next().inputFileNum()).isEqualTo(2);
+ }
+
+ @Test
+ void testEmptyTable() {
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(REWRITE_ALL);
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(table.currentSnapshot()).as("Table must be empty").isNull();
+ assertThat(plan.totalGroupCount()).isZero();
+ }
+
+ @ParameterizedTest
+ @EnumSource(
+ value = RewriteJobOrder.class,
+ names = {"FILES_DESC", "FILES_ASC", "BYTES_DESC", "BYTES_ASC"})
+ void testJobOrder(RewriteJobOrder order) {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true", REWRITE_JOB_ORDER, order.name()));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups.stream().map(group -> group.info().partition()).collect(Collectors.toList()))
+ .isEqualTo(EXPECTED.get(order));
+ assertThat(plan.totalGroupCount()).isEqualTo(3);
+ EXPECTED.get(order).forEach(s -> assertThat(plan.groupsInPartition(s)).isEqualTo(1));
+ }
+
+ @Test
+ void testMaxGroupSize() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.MAX_FILE_GROUP_SIZE_BYTES,
+ "10"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.totalGroupCount()).isEqualTo(6);
+ assertThat(plan.groupsInPartition(FILE_1.partition())).isEqualTo(3);
+ assertThat(plan.groupsInPartition(FILE_4.partition())).isEqualTo(2);
+ assertThat(plan.groupsInPartition(FILE_6.partition())).isEqualTo(1);
+ }
+
+ @Test
+ void testFilter() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner =
+ new BinPackRewriteFileGroupPlanner(
+ table,
+ Expressions.or(
+ Expressions.equal(Expressions.bucket("data", 16), 0),
+ Expressions.equal(Expressions.bucket("data", 16), 2)));
+ planner.init(REWRITE_ALL);
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(plan.totalGroupCount()).isEqualTo(2);
+ assertThat(groups).hasSize(2);
+ assertThat(groups.stream().mapToLong(RewriteFileGroup::inputFileNum).sum()).isEqualTo(4);
+ }
+
+ @Test
+ void testMaxOutputFileSize() {
+ addFiles();
+ int targetFileSize = 10;
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.TARGET_FILE_SIZE_BYTES,
+ String.valueOf(targetFileSize)));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.groups().iterator().next().maxOutputFileSize())
+ .isGreaterThan(targetFileSize)
+ .isLessThan((long) (targetFileSize * MAX_FILE_SIZE_DEFAULT_RATIO));
+ }
+
+ @Test
+ void testExpectedOutputFiles() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.TARGET_FILE_SIZE_BYTES,
+ "21"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups).hasSize(3);
+ for (RewriteFileGroup group : groups) {
+ if (FILE_1.partition().equals(group.info().partition())) {
+ assertThat(group.expectedOutputFiles()).isEqualTo(2);
+ } else if (FILE_4.partition().equals(group.info().partition())) {
+ assertThat(group.expectedOutputFiles()).isEqualTo(1);
+ } else if (FILE_6.partition().equals(group.info().partition())) {
+ assertThat(group.expectedOutputFiles()).isEqualTo(3);
+ } else {
+ throw new IllegalStateException("Unexpected partition: " + group.info().partition());
+ }
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testOutputSpec(boolean specific) {
+ addFiles();
+ int oldSpecId = table.spec().specId();
+ table.updateSpec().removeField("data_bucket").commit();
+ table.newAppend().appendFile(newDataFile("", 10)).commit();
+ table.refresh();
+ int newSpecId = table.spec().specId();
+
+ Map<String, String> options = Maps.newHashMap(REWRITE_ALL);
+ if (specific) {
+ options.put(RewriteDataFiles.OUTPUT_SPEC_ID, String.valueOf(oldSpecId));
+ }
+
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(options);
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.groups().iterator().next().outputSpecId())
+ .isEqualTo(specific ? oldSpecId : newSpecId);
+ }
+
+ @Test
+ void testValidOptions() {
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+
+ assertThat(planner.validOptions())
+ .as("Planner must report all supported options")
+ .isEqualTo(
+ ImmutableSet.of(
+ BinPackRewriteFileGroupPlanner.TARGET_FILE_SIZE_BYTES,
+ BinPackRewriteFileGroupPlanner.MIN_FILE_SIZE_BYTES,
+ BinPackRewriteFileGroupPlanner.MAX_FILE_SIZE_BYTES,
+ BinPackRewriteFileGroupPlanner.MIN_INPUT_FILES,
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ BinPackRewriteFileGroupPlanner.MAX_FILE_GROUP_SIZE_BYTES,
+ BinPackRewriteFileGroupPlanner.DELETE_FILE_THRESHOLD,
+ BinPackRewriteFileGroupPlanner.DELETE_RATIO_THRESHOLD,
+ RewriteDataFiles.REWRITE_JOB_ORDER));
+ }
+
+ @Test
+ void testInvalidOption() {
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+
+ Map<String, String> invalidRewriteJobOrderOptions =
+ ImmutableMap.of(RewriteDataFiles.REWRITE_JOB_ORDER, "foo");
+ assertThatThrownBy(() -> planner.init(invalidRewriteJobOrderOptions))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid rewrite job order name: foo");
+
+ Map<String, String> invalidOutputSpecIdOptions =
+ ImmutableMap.of(RewriteDataFiles.OUTPUT_SPEC_ID, String.valueOf(1234));
+ assertThatThrownBy(() -> planner.init(invalidOutputSpecIdOptions))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage(
+ "Cannot use output spec id 1234 because the table does not contain a reference to this spec-id.");
+
+ Map<String, String> invalidDeleteFileThresholdOptions =
+ ImmutableMap.of(BinPackRewriteFileGroupPlanner.DELETE_FILE_THRESHOLD, "-1");
+ assertThatThrownBy(() -> planner.init(invalidDeleteFileThresholdOptions))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("'delete-file-threshold' is set to -1 but must be >= 0");
+
+ Map<String, String> negativeDeleteRatioThresholdOptions =
+ ImmutableMap.of(BinPackRewriteFileGroupPlanner.DELETE_RATIO_THRESHOLD, "-1");
+ assertThatThrownBy(() -> planner.init(negativeDeleteRatioThresholdOptions))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("'delete-ratio-threshold' is set to -1.0 but must be > 0");
+
+ Map<String, String> invalidDeleteRatioThresholdOptions =
+ ImmutableMap.of(BinPackRewriteFileGroupPlanner.DELETE_RATIO_THRESHOLD, "127");
+ assertThatThrownBy(() -> planner.init(invalidDeleteRatioThresholdOptions))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("'delete-ratio-threshold' is set to 127.0 but must be <= 1");
+ }
+
+ @Test
+ void testSizeFiltering() {
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ FileScanTask tooSmallTask = new MockFileScanTask(100L);
+ FileScanTask optimal = new MockFileScanTask(450);
+ FileScanTask tooBigTask = new MockFileScanTask(1000L);
+ List<FileScanTask> tasks = ImmutableList.of(tooSmallTask, optimal, tooBigTask);
+
+ Map<String, String> options =
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.MIN_FILE_SIZE_BYTES, "250",
+ BinPackRewriteFileGroupPlanner.TARGET_FILE_SIZE_BYTES, "500",
+ BinPackRewriteFileGroupPlanner.MAX_FILE_SIZE_BYTES, "750");
+ planner.init(options);
+
+ Iterable<List<FileScanTask>> groups = planner.planFileGroups(tasks);
+ assertThat(groups).as("Must have 1 group").hasSize(1);
+ List<FileScanTask> group = Iterables.getOnlyElement(groups);
+ assertThat(group).as("Must rewrite 2 files").hasSize(2);
+ assertThat(group).containsExactlyInAnyOrder(tooSmallTask, tooBigTask);
+ }
+
+ @Test
+ void testGroupWithEnoughFiles() {
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ List<FileScanTask> tasks =
+ ImmutableList.of(
+ new MockFileScanTask(100L),
+ new MockFileScanTask(100L),
+ new MockFileScanTask(100L),
+ new MockFileScanTask(100L));
+
+ Map<String, String> options =
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.MIN_INPUT_FILES, "3",
+ BinPackRewriteFileGroupPlanner.MIN_FILE_SIZE_BYTES, "150",
+ BinPackRewriteFileGroupPlanner.TARGET_FILE_SIZE_BYTES, "1000",
+ BinPackRewriteFileGroupPlanner.MAX_FILE_SIZE_BYTES, "5000");
+ planner.init(options);
+
+ Iterable<List<FileScanTask>> groups = planner.planFileGroups(tasks);
+ assertThat(groups).as("Must have 1 group").hasSize(1);
+ List<FileScanTask> group = Iterables.getOnlyElement(groups);
+ assertThat(group).as("Must rewrite 4 files").hasSize(4);
+ }
+
+ @Test
+ void testGroupWithEnoughData() {
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ List<FileScanTask> tasks =
+ ImmutableList.of(
+ new MockFileScanTask(100L), new MockFileScanTask(100L), new MockFileScanTask(100L));
+
+ Map<String, String> options =
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.MIN_FILE_SIZE_BYTES, "200",
+ BinPackRewriteFileGroupPlanner.TARGET_FILE_SIZE_BYTES, "250",
Review Comment:
should this be larger than 300 so the the MIN_FILE_SIZE_BYTES is the trigger condition?
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2016087097
##########
core/src/main/java/org/apache/iceberg/actions/BinPackRewriteFileGroupPlanner.java:
##########
@@ -0,0 +1,310 @@
+/*
+ * 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.actions;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.actions.RewriteDataFiles.FileGroupInfo;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.ContentFileUtil;
+import org.apache.iceberg.util.PropertyUtil;
+import org.apache.iceberg.util.StructLikeMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Groups specified data files in the {@link Table} into {@link RewriteFileGroup}s. The files are
+ * grouped by partitions based on their size using fix sized bins. Extends {@link
+ * SizeBasedFileRewritePlanner} with delete file number and delete ratio thresholds and job {@link
+ * RewriteDataFiles#REWRITE_JOB_ORDER} handling.
+ */
+public class BinPackRewriteFileGroupPlanner
+ extends SizeBasedFileRewritePlanner<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> {
+ /**
+ * The minimum number of deletes that needs to be associated with a data file for it to be
+ * considered for rewriting. If a data file has this number of deletes or more, it will be
+ * rewritten regardless of its file size determined by {@link #MIN_FILE_SIZE_BYTES} and {@link
+ * #MAX_FILE_SIZE_BYTES}. If a file group contains a file that satisfies this condition, the file
+ * group will be rewritten regardless of the number of files in the file group determined by
+ * {@link #MIN_INPUT_FILES}.
+ *
+ * <p>Defaults to Integer.MAX_VALUE, which means this feature is not enabled by default.
+ */
+ public static final String DELETE_FILE_THRESHOLD = "delete-file-threshold";
Review Comment:
While I agree with you, I also agree with you that we don't want to touch this in this PR (and probably not even later)
##########
core/src/main/java/org/apache/iceberg/actions/BinPackRewriteFileGroupPlanner.java:
##########
@@ -0,0 +1,310 @@
+/*
+ * 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.actions;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.actions.RewriteDataFiles.FileGroupInfo;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.ContentFileUtil;
+import org.apache.iceberg.util.PropertyUtil;
+import org.apache.iceberg.util.StructLikeMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Groups specified data files in the {@link Table} into {@link RewriteFileGroup}s. The files are
+ * grouped by partitions based on their size using fix sized bins. Extends {@link
+ * SizeBasedFileRewritePlanner} with delete file number and delete ratio thresholds and job {@link
+ * RewriteDataFiles#REWRITE_JOB_ORDER} handling.
+ */
+public class BinPackRewriteFileGroupPlanner
+ extends SizeBasedFileRewritePlanner<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> {
+ /**
+ * The minimum number of deletes that needs to be associated with a data file for it to be
+ * considered for rewriting. If a data file has this number of deletes or more, it will be
+ * rewritten regardless of its file size determined by {@link #MIN_FILE_SIZE_BYTES} and {@link
+ * #MAX_FILE_SIZE_BYTES}. If a file group contains a file that satisfies this condition, the file
+ * group will be rewritten regardless of the number of files in the file group determined by
+ * {@link #MIN_INPUT_FILES}.
+ *
+ * <p>Defaults to Integer.MAX_VALUE, which means this feature is not enabled by default.
+ */
+ public static final String DELETE_FILE_THRESHOLD = "delete-file-threshold";
+
+ public static final int DELETE_FILE_THRESHOLD_DEFAULT = Integer.MAX_VALUE;
+
+ /**
+ * The minimum deletion ratio that needs to be associated with a data file for it to be considered
+ * for rewriting. If the deletion ratio of a data file is greater than or equal to this value, it
+ * will be rewritten regardless of its file size determined by {@link #MIN_FILE_SIZE_BYTES} and
+ * {@link #MAX_FILE_SIZE_BYTES}. If a file group contains a file that satisfies this condition,
+ * the file group will be rewritten regardless of the number of files in the file group determined
+ * by {@link #MIN_INPUT_FILES}.
+ *
+ * <p>Defaults to 0.3, which means that if the deletion ratio of a file reaches or exceeds 30%, it
+ * may trigger the rewriting operation.
+ */
+ public static final String DELETE_RATIO_THRESHOLD = "delete-ratio-threshold";
+
+ public static final double DELETE_RATIO_THRESHOLD_DEFAULT = 0.3;
+
+ private static final Logger LOG = LoggerFactory.getLogger(BinPackRewriteFileGroupPlanner.class);
+
+ private final Expression filter;
+ private final Long snapshotId;
+ private final boolean caseSensitive;
+
+ private int deleteFileThreshold;
+ private double deleteRatioThreshold;
+ private RewriteJobOrder rewriteJobOrder;
+
+ public BinPackRewriteFileGroupPlanner(Table table) {
+ this(table, Expressions.alwaysTrue());
+ }
+
+ public BinPackRewriteFileGroupPlanner(Table table, Expression filter) {
+ this(
+ table,
+ filter,
+ table.currentSnapshot() != null ? table.currentSnapshot().snapshotId() : null,
+ false);
+ }
+
+ /**
+ * Creates the planner for the given table.
+ *
+ * @param table to plan for
+ * @param filter used to remove files from the plan
+ * @param snapshotId used as a basis for planning - should be used as starting snapshot id at
Review Comment:
Applied
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2016081277
##########
core/src/main/java/org/apache/iceberg/actions/RewriteFileGroupPlanner.java:
##########
@@ -0,0 +1,310 @@
+/*
+ * 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.actions;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.actions.RewriteDataFiles.FileGroupInfo;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.ContentFileUtil;
+import org.apache.iceberg.util.PropertyUtil;
+import org.apache.iceberg.util.StructLikeMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Groups specified data files in the {@link Table} into {@link RewriteFileGroup}s. The files are
+ * grouped by partitions based on their size using fix sized bins. Extends {@link
+ * SizeBasedFileRewritePlanner} with delete file number and delete ratio thresholds and job {@link
+ * RewriteDataFiles#REWRITE_JOB_ORDER} handling.
+ */
+public class RewriteFileGroupPlanner
+ extends SizeBasedFileRewritePlanner<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> {
+ /**
+ * The minimum number of deletes that needs to be associated with a data file for it to be
+ * considered for rewriting. If a data file has this number of deletes or more, it will be
+ * rewritten regardless of its file size determined by {@link #MIN_FILE_SIZE_BYTES} and {@link
+ * #MAX_FILE_SIZE_BYTES}. If a file group contains a file that satisfies this condition, the file
+ * group will be rewritten regardless of the number of files in the file group determined by
+ * {@link #MIN_INPUT_FILES}.
+ *
+ * <p>Defaults to Integer.MAX_VALUE, which means this feature is not enabled by default.
+ */
+ public static final String DELETE_FILE_THRESHOLD = "delete-file-threshold";
+
+ public static final int DELETE_FILE_THRESHOLD_DEFAULT = Integer.MAX_VALUE;
+
+ /**
+ * The minimum deletion ratio that needs to be associated with a data file for it to be considered
Review Comment:
Applied @stevenzwu's suggestions
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary merged PR #12493:
URL: https://github.com/apache/iceberg/pull/12493
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2012931655
##########
core/src/main/java/org/apache/iceberg/actions/RewriteFileGroupPlanner.java:
##########
@@ -0,0 +1,310 @@
+/*
+ * 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.actions;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.actions.RewriteDataFiles.FileGroupInfo;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.ContentFileUtil;
+import org.apache.iceberg.util.PropertyUtil;
+import org.apache.iceberg.util.StructLikeMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Groups specified data files in the {@link Table} into {@link RewriteFileGroup}s. The files are
+ * grouped by partitions based on their size using fix sized bins. Extends {@link
+ * SizeBasedFileRewritePlanner} with delete file number and delete ratio thresholds and job {@link
+ * RewriteDataFiles#REWRITE_JOB_ORDER} handling.
+ */
+public class RewriteFileGroupPlanner
Review Comment:
Not sure I understand the extends here, RewriteFileGroupPlanner sounds more general than SizeBasedFileRewritePlanner I may be missing something i'll try to get back to this after I read the rest of the 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] Core: FileRewritePlanner implementation [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2015149619
##########
core/src/test/java/org/apache/iceberg/actions/TestBinPackRewriteFileGroupPlanner.java:
##########
@@ -0,0 +1,468 @@
+/*
+ * 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.actions;
+
+import static org.apache.iceberg.actions.BinPackRewriteFileGroupPlanner.MAX_FILE_SIZE_DEFAULT_RATIO;
+import static org.apache.iceberg.actions.RewriteDataFiles.REWRITE_JOB_ORDER;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.File;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.stream.Collectors;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.MockFileScanTask;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.TestBase;
+import org.apache.iceberg.TestTables;
+import org.apache.iceberg.actions.RewriteDataFiles.FileGroupInfo;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class TestBinPackRewriteFileGroupPlanner {
+ private static final Map<String, String> REWRITE_ALL =
+ ImmutableMap.of(BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true");
+
+ private static final DataFile FILE_1 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_2 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_3 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_4 = newDataFile("data_bucket=1", 11);
+ private static final DataFile FILE_5 = newDataFile("data_bucket=1", 11);
+ private static final DataFile FILE_6 = newDataFile("data_bucket=2", 50);
+
+ private static final Map<RewriteJobOrder, List<StructLike>> EXPECTED =
+ ImmutableMap.of(
+ RewriteJobOrder.FILES_DESC,
+ ImmutableList.of(FILE_1.partition(), FILE_4.partition(), FILE_6.partition()),
+ RewriteJobOrder.FILES_ASC,
+ ImmutableList.of(FILE_6.partition(), FILE_4.partition(), FILE_1.partition()),
+ RewriteJobOrder.BYTES_DESC,
+ ImmutableList.of(FILE_6.partition(), FILE_1.partition(), FILE_4.partition()),
+ RewriteJobOrder.BYTES_ASC,
+ ImmutableList.of(FILE_4.partition(), FILE_1.partition(), FILE_6.partition()));
+
+ @TempDir private File tableDir = null;
+ private TestTables.TestTable table = null;
+
+ @BeforeEach
+ public void setupTable() throws Exception {
+ this.table = TestTables.create(tableDir, "test", TestBase.SCHEMA, TestBase.SPEC, 3);
+ }
+
+ @AfterEach
+ public void cleanupTables() {
+ TestTables.clearTables();
+ }
+
+ @Test
+ void testPartitionedTable() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(ImmutableMap.of(BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups).hasSize(3);
+ assertThat(groups.stream().mapToInt(RewriteGroupBase::inputFileNum).sum()).isEqualTo(6);
+ }
+
+ @Test
+ void testUnpartitionedTable() {
+ table.updateSpec().removeField("data_bucket").commit();
+ table.refresh();
+ table
+ .newAppend()
+ .appendFile(newDataFile("", 10))
+ .appendFile(newDataFile("", 20))
+ .appendFile(newDataFile("", 30))
+ .commit();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.MIN_INPUT_FILES,
+ "1",
+ BinPackRewriteFileGroupPlanner.MIN_FILE_SIZE_BYTES,
+ "30"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.totalGroupCount()).isEqualTo(1);
+ assertThat(plan.groups().iterator().next().inputFileNum()).isEqualTo(2);
+ }
+
+ @Test
+ void testEmptyTable() {
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(REWRITE_ALL);
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(table.currentSnapshot()).as("Table must be empty").isNull();
+ assertThat(plan.totalGroupCount()).isZero();
+ }
+
+ @ParameterizedTest
+ @EnumSource(
+ value = RewriteJobOrder.class,
+ names = {"FILES_DESC", "FILES_ASC", "BYTES_DESC", "BYTES_ASC"})
+ void testJobOrder(RewriteJobOrder order) {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true", REWRITE_JOB_ORDER, order.name()));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups.stream().map(group -> group.info().partition()).collect(Collectors.toList()))
+ .isEqualTo(EXPECTED.get(order));
+ assertThat(plan.totalGroupCount()).isEqualTo(3);
+ EXPECTED.get(order).forEach(s -> assertThat(plan.groupsInPartition(s)).isEqualTo(1));
+ }
+
+ @Test
+ void testMaxGroupSize() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.MAX_FILE_GROUP_SIZE_BYTES,
+ "10"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.totalGroupCount()).isEqualTo(6);
+ assertThat(plan.groupsInPartition(FILE_1.partition())).isEqualTo(3);
+ assertThat(plan.groupsInPartition(FILE_4.partition())).isEqualTo(2);
+ assertThat(plan.groupsInPartition(FILE_6.partition())).isEqualTo(1);
+ }
+
+ @Test
+ void testFilter() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner =
+ new BinPackRewriteFileGroupPlanner(
+ table,
+ Expressions.or(
+ Expressions.equal(Expressions.bucket("data", 16), 0),
+ Expressions.equal(Expressions.bucket("data", 16), 2)));
+ planner.init(REWRITE_ALL);
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(plan.totalGroupCount()).isEqualTo(2);
+ assertThat(groups).hasSize(2);
+ assertThat(groups.stream().mapToLong(RewriteFileGroup::inputFileNum).sum()).isEqualTo(4);
+ }
+
+ @Test
+ void testMaxOutputFileSize() {
+ addFiles();
+ int targetFileSize = 10;
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.TARGET_FILE_SIZE_BYTES,
+ String.valueOf(targetFileSize)));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.groups().iterator().next().maxOutputFileSize())
Review Comment:
nit: validate against every group?
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2015125611
##########
core/src/main/java/org/apache/iceberg/actions/RewritePositionDeletesGroupPlanner.java:
##########
@@ -0,0 +1,221 @@
+/*
+ * 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.actions;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.MetadataTableType;
+import org.apache.iceberg.MetadataTableUtils;
+import org.apache.iceberg.Partitioning;
+import org.apache.iceberg.PositionDeletesScanTask;
+import org.apache.iceberg.PositionDeletesTable;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.actions.RewritePositionDeleteFiles.FileGroupInfo;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.PartitionUtil;
+import org.apache.iceberg.util.PropertyUtil;
+import org.apache.iceberg.util.StructLikeMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Groups specified delete files in the {@link Table} into {@link RewritePositionDeletesGroup}s.
+ * These will be grouped by partitions based on their size using fix sized bins. Extends the {@link
+ * SizeBasedFileRewritePlanner} with {@link RewritePositionDeleteFiles#REWRITE_JOB_ORDER} handling.
+ */
+public class RewritePositionDeletesGroupPlanner
+ extends SizeBasedFileRewritePlanner<
+ FileGroupInfo, PositionDeletesScanTask, DeleteFile, RewritePositionDeletesGroup> {
+ private static final Logger LOG =
+ LoggerFactory.getLogger(RewritePositionDeletesGroupPlanner.class);
+
+ private final Expression filter;
+ private final boolean caseSensitive;
+ private RewriteJobOrder rewriteJobOrder;
+
+ public RewritePositionDeletesGroupPlanner(Table table) {
+ this(table, Expressions.alwaysTrue(), false);
+ }
+
+ /**
+ * Creates the planner for the given table.
+ *
+ * @param table to plan for
+ * @param filter used to remove files from the plan
+ * @param caseSensitive property used for scanning
+ */
+ public RewritePositionDeletesGroupPlanner(Table table, Expression filter, boolean caseSensitive) {
+ super(table);
+ this.caseSensitive = caseSensitive;
+ this.filter = filter;
+ }
+
+ @Override
+ public Set<String> validOptions() {
+ return ImmutableSet.<String>builder()
+ .addAll(super.validOptions())
+ .add(RewritePositionDeleteFiles.REWRITE_JOB_ORDER)
+ .build();
+ }
+
+ @Override
+ public void init(Map<String, String> options) {
+ super.init(options);
+ this.rewriteJobOrder =
+ RewriteJobOrder.fromName(
+ PropertyUtil.propertyAsString(
+ options,
+ RewritePositionDeleteFiles.REWRITE_JOB_ORDER,
+ RewritePositionDeleteFiles.REWRITE_JOB_ORDER_DEFAULT));
+ }
+
+ @Override
+ public FileRewritePlan<
+ FileGroupInfo, PositionDeletesScanTask, DeleteFile, RewritePositionDeletesGroup>
+ plan() {
+ StructLikeMap<List<List<PositionDeletesScanTask>>> plan = planFileGroups();
+ RewriteExecutionContext ctx = new RewriteExecutionContext();
+ Stream<RewritePositionDeletesGroup> groups =
+ plan.entrySet().stream()
+ .filter(e -> !e.getValue().isEmpty())
+ .flatMap(
+ e -> {
+ StructLike partition = e.getKey();
+ List<List<PositionDeletesScanTask>> scanGroups = e.getValue();
+ return scanGroups.stream()
+ .map(
+ tasks -> {
+ long inputSize = inputSize(tasks);
+ return newRewriteGroup(
+ ctx,
+ partition,
+ tasks,
+ inputSplitSize(inputSize),
+ expectedOutputFiles(inputSize));
+ });
+ })
+ .sorted(RewritePositionDeletesGroup.comparator(rewriteJobOrder));
+ Map<StructLike, Integer> groupsInPartition = plan.transformValues(List::size);
+ int totalGroupCount = groupsInPartition.values().stream().reduce(Integer::sum).orElse(0);
+ return new FileRewritePlan<>(
+ CloseableIterable.of(groups.collect(Collectors.toList())),
+ totalGroupCount,
+ groupsInPartition);
+ }
+
+ @Override
+ protected Iterable<PositionDeletesScanTask> filterFiles(Iterable<PositionDeletesScanTask> tasks) {
+ return Iterables.filter(tasks, this::wronglySized);
+ }
+
+ @Override
+ protected Iterable<List<PositionDeletesScanTask>> filterFileGroups(
+ List<List<PositionDeletesScanTask>> groups) {
+ return Iterables.filter(
+ groups, group -> enoughInputFiles(group) || enoughContent(group) || tooMuchContent(group));
+ }
+
+ @Override
+ protected long defaultTargetFileSize() {
+ return PropertyUtil.propertyAsLong(
+ table().properties(),
+ TableProperties.DELETE_TARGET_FILE_SIZE_BYTES,
+ TableProperties.DELETE_TARGET_FILE_SIZE_BYTES_DEFAULT);
+ }
+
+ private StructLikeMap<List<List<PositionDeletesScanTask>>> planFileGroups() {
+ Table deletesTable =
+ MetadataTableUtils.createMetadataTableInstance(table(), MetadataTableType.POSITION_DELETES);
+ CloseableIterable<PositionDeletesScanTask> fileTasks = planFiles(deletesTable);
+
+ try {
+ Types.StructType partitionType = Partitioning.partitionType(deletesTable);
+ StructLikeMap<List<PositionDeletesScanTask>> fileTasksByPartition =
+ groupByPartition(partitionType, fileTasks);
+ return fileTasksByPartition.transformValues(
+ tasks -> ImmutableList.copyOf(planFileGroups(tasks)));
+ } finally {
+ try {
+ fileTasks.close();
+ } catch (IOException io) {
+ LOG.error("Cannot properly close file iterable while planning for rewrite", io);
+ }
+ }
+ }
+
+ private CloseableIterable<PositionDeletesScanTask> planFiles(Table deletesTable) {
+ PositionDeletesTable.PositionDeletesBatchScan scan =
+ (PositionDeletesTable.PositionDeletesBatchScan) deletesTable.newBatchScan();
+ return CloseableIterable.transform(
+ scan.baseTableFilter(filter).caseSensitive(caseSensitive).ignoreResiduals().planFiles(),
+ PositionDeletesScanTask.class::cast);
+ }
+
+ private StructLikeMap<List<PositionDeletesScanTask>> groupByPartition(
Review Comment:
some of the methods in this class look similar to the code in the `BinPackRewriteFilePlanner`. can the code be pushed to the base class? e.g. `plan()`, most of `planFileGroups`, etc.
In this specific method, I guess `PartitionUtil.coercePartition` is probably different
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2013727684
##########
core/src/main/java/org/apache/iceberg/actions/RewriteFileGroupPlanner.java:
##########
@@ -0,0 +1,310 @@
+/*
+ * 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.actions;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.actions.RewriteDataFiles.FileGroupInfo;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.ContentFileUtil;
+import org.apache.iceberg.util.PropertyUtil;
+import org.apache.iceberg.util.StructLikeMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Groups specified data files in the {@link Table} into {@link RewriteFileGroup}s. The files are
+ * grouped by partitions based on their size using fix sized bins. Extends {@link
+ * SizeBasedFileRewritePlanner} with delete file number and delete ratio thresholds and job {@link
+ * RewriteDataFiles#REWRITE_JOB_ORDER} handling.
+ */
+public class RewriteFileGroupPlanner
+ extends SizeBasedFileRewritePlanner<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> {
+ /**
+ * The minimum number of deletes that needs to be associated with a data file for it to be
+ * considered for rewriting. If a data file has this number of deletes or more, it will be
+ * rewritten regardless of its file size determined by {@link #MIN_FILE_SIZE_BYTES} and {@link
+ * #MAX_FILE_SIZE_BYTES}. If a file group contains a file that satisfies this condition, the file
+ * group will be rewritten regardless of the number of files in the file group determined by
+ * {@link #MIN_INPUT_FILES}.
+ *
+ * <p>Defaults to Integer.MAX_VALUE, which means this feature is not enabled by default.
+ */
+ public static final String DELETE_FILE_THRESHOLD = "delete-file-threshold";
+
+ public static final int DELETE_FILE_THRESHOLD_DEFAULT = Integer.MAX_VALUE;
+
+ /**
+ * The minimum deletion ratio that needs to be associated with a data file for it to be considered
+ * for rewriting. If the deletion ratio of a data file is greater than or equal to this value, it
+ * will be rewritten regardless of its file size determined by {@link #MIN_FILE_SIZE_BYTES} and
+ * {@link #MAX_FILE_SIZE_BYTES}. If a file group contains a file that satisfies this condition,
+ * the file group will be rewritten regardless of the number of files in the file group determined
+ * by {@link #MIN_INPUT_FILES}.
+ *
+ * <p>Defaults to 0.3, which means that if the deletion ratio of a file reaches or exceeds 30%, it
+ * may trigger the rewriting operation.
+ */
+ public static final String DELETE_RATIO_THRESHOLD = "delete-ratio-threshold";
+
+ public static final double DELETE_RATIO_THRESHOLD_DEFAULT = 0.3;
+
+ private static final Logger LOG = LoggerFactory.getLogger(RewriteFileGroupPlanner.class);
+
+ private final Expression filter;
+ private final Long snapshotId;
+ private final boolean caseSensitive;
+
+ private int deleteFileThreshold;
+ private double deleteRatioThreshold;
+ private RewriteJobOrder rewriteJobOrder;
+
+ public RewriteFileGroupPlanner(Table table) {
+ this(table, Expressions.alwaysTrue());
+ }
+
+ public RewriteFileGroupPlanner(Table table, Expression filter) {
+ this(
+ table,
+ filter,
+ table.currentSnapshot() != null ? table.currentSnapshot().snapshotId() : null,
+ false);
+ }
+
+ /**
+ * Creates the planner for the given table.
+ *
+ * @param table to plan for
+ * @param filter used to remove files from the plan
+ * @param snapshotId used as a basis for planning - should be used as starting snapshot id at
+ * commit time when replacing the files
+ * @param caseSensitive property used for scanning
+ */
+ public RewriteFileGroupPlanner(
+ Table table, Expression filter, Long snapshotId, boolean caseSensitive) {
+ super(table);
+ this.filter = filter;
+ this.snapshotId = snapshotId;
+ this.caseSensitive = caseSensitive;
+ }
+
+ @Override
+ public Set<String> validOptions() {
+ return ImmutableSet.<String>builder()
+ .addAll(super.validOptions())
+ .add(DELETE_FILE_THRESHOLD)
+ .add(DELETE_RATIO_THRESHOLD)
+ .add(RewriteDataFiles.REWRITE_JOB_ORDER)
+ .build();
+ }
+
+ @Override
+ public void init(Map<String, String> options) {
+ super.init(options);
+ this.deleteFileThreshold = deleteFileThreshold(options);
+ this.deleteRatioThreshold = deleteRatioThreshold(options);
+ this.rewriteJobOrder =
+ RewriteJobOrder.fromName(
+ PropertyUtil.propertyAsString(
+ options,
+ RewriteDataFiles.REWRITE_JOB_ORDER,
Review Comment:
The plan currently returns the jobs in order in a `CloseableIterable`.
We can push the ordering to the user, but this seems something we can reuse between Spark and Flink, so I decided to keep it here.
Your thoughts?
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2015152999
##########
core/src/test/java/org/apache/iceberg/actions/TestBinPackRewriteFileGroupPlanner.java:
##########
@@ -0,0 +1,468 @@
+/*
+ * 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.actions;
+
+import static org.apache.iceberg.actions.BinPackRewriteFileGroupPlanner.MAX_FILE_SIZE_DEFAULT_RATIO;
+import static org.apache.iceberg.actions.RewriteDataFiles.REWRITE_JOB_ORDER;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.File;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.stream.Collectors;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.MockFileScanTask;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.TestBase;
+import org.apache.iceberg.TestTables;
+import org.apache.iceberg.actions.RewriteDataFiles.FileGroupInfo;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class TestBinPackRewriteFileGroupPlanner {
+ private static final Map<String, String> REWRITE_ALL =
+ ImmutableMap.of(BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true");
+
+ private static final DataFile FILE_1 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_2 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_3 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_4 = newDataFile("data_bucket=1", 11);
+ private static final DataFile FILE_5 = newDataFile("data_bucket=1", 11);
+ private static final DataFile FILE_6 = newDataFile("data_bucket=2", 50);
+
+ private static final Map<RewriteJobOrder, List<StructLike>> EXPECTED =
+ ImmutableMap.of(
+ RewriteJobOrder.FILES_DESC,
+ ImmutableList.of(FILE_1.partition(), FILE_4.partition(), FILE_6.partition()),
+ RewriteJobOrder.FILES_ASC,
+ ImmutableList.of(FILE_6.partition(), FILE_4.partition(), FILE_1.partition()),
+ RewriteJobOrder.BYTES_DESC,
+ ImmutableList.of(FILE_6.partition(), FILE_1.partition(), FILE_4.partition()),
+ RewriteJobOrder.BYTES_ASC,
+ ImmutableList.of(FILE_4.partition(), FILE_1.partition(), FILE_6.partition()));
+
+ @TempDir private File tableDir = null;
+ private TestTables.TestTable table = null;
+
+ @BeforeEach
+ public void setupTable() throws Exception {
+ this.table = TestTables.create(tableDir, "test", TestBase.SCHEMA, TestBase.SPEC, 3);
+ }
+
+ @AfterEach
+ public void cleanupTables() {
+ TestTables.clearTables();
+ }
+
+ @Test
+ void testPartitionedTable() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(ImmutableMap.of(BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups).hasSize(3);
+ assertThat(groups.stream().mapToInt(RewriteGroupBase::inputFileNum).sum()).isEqualTo(6);
+ }
+
+ @Test
+ void testUnpartitionedTable() {
+ table.updateSpec().removeField("data_bucket").commit();
+ table.refresh();
+ table
+ .newAppend()
+ .appendFile(newDataFile("", 10))
+ .appendFile(newDataFile("", 20))
+ .appendFile(newDataFile("", 30))
+ .commit();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.MIN_INPUT_FILES,
+ "1",
+ BinPackRewriteFileGroupPlanner.MIN_FILE_SIZE_BYTES,
+ "30"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.totalGroupCount()).isEqualTo(1);
+ assertThat(plan.groups().iterator().next().inputFileNum()).isEqualTo(2);
+ }
+
+ @Test
+ void testEmptyTable() {
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(REWRITE_ALL);
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(table.currentSnapshot()).as("Table must be empty").isNull();
+ assertThat(plan.totalGroupCount()).isZero();
+ }
+
+ @ParameterizedTest
+ @EnumSource(
+ value = RewriteJobOrder.class,
+ names = {"FILES_DESC", "FILES_ASC", "BYTES_DESC", "BYTES_ASC"})
+ void testJobOrder(RewriteJobOrder order) {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true", REWRITE_JOB_ORDER, order.name()));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups.stream().map(group -> group.info().partition()).collect(Collectors.toList()))
+ .isEqualTo(EXPECTED.get(order));
+ assertThat(plan.totalGroupCount()).isEqualTo(3);
+ EXPECTED.get(order).forEach(s -> assertThat(plan.groupsInPartition(s)).isEqualTo(1));
+ }
+
+ @Test
+ void testMaxGroupSize() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.MAX_FILE_GROUP_SIZE_BYTES,
+ "10"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.totalGroupCount()).isEqualTo(6);
+ assertThat(plan.groupsInPartition(FILE_1.partition())).isEqualTo(3);
+ assertThat(plan.groupsInPartition(FILE_4.partition())).isEqualTo(2);
+ assertThat(plan.groupsInPartition(FILE_6.partition())).isEqualTo(1);
+ }
+
+ @Test
+ void testFilter() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner =
+ new BinPackRewriteFileGroupPlanner(
+ table,
+ Expressions.or(
+ Expressions.equal(Expressions.bucket("data", 16), 0),
+ Expressions.equal(Expressions.bucket("data", 16), 2)));
+ planner.init(REWRITE_ALL);
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(plan.totalGroupCount()).isEqualTo(2);
+ assertThat(groups).hasSize(2);
+ assertThat(groups.stream().mapToLong(RewriteFileGroup::inputFileNum).sum()).isEqualTo(4);
+ }
+
+ @Test
+ void testMaxOutputFileSize() {
+ addFiles();
+ int targetFileSize = 10;
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.TARGET_FILE_SIZE_BYTES,
+ String.valueOf(targetFileSize)));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.groups().iterator().next().maxOutputFileSize())
+ .isGreaterThan(targetFileSize)
+ .isLessThan((long) (targetFileSize * MAX_FILE_SIZE_DEFAULT_RATIO));
+ }
+
+ @Test
+ void testExpectedOutputFiles() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.TARGET_FILE_SIZE_BYTES,
+ "21"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups).hasSize(3);
+ for (RewriteFileGroup group : groups) {
+ if (FILE_1.partition().equals(group.info().partition())) {
+ assertThat(group.expectedOutputFiles()).isEqualTo(2);
+ } else if (FILE_4.partition().equals(group.info().partition())) {
+ assertThat(group.expectedOutputFiles()).isEqualTo(1);
Review Comment:
is this the case to avoid the 2nd small remainder file? otherwise, group size of 22 is larger than the target size of 21.
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2015046752
##########
core/src/main/java/org/apache/iceberg/actions/BinPackRewriteFileGroupPlanner.java:
##########
@@ -0,0 +1,310 @@
+/*
+ * 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.actions;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.actions.RewriteDataFiles.FileGroupInfo;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.ContentFileUtil;
+import org.apache.iceberg.util.PropertyUtil;
+import org.apache.iceberg.util.StructLikeMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Groups specified data files in the {@link Table} into {@link RewriteFileGroup}s. The files are
+ * grouped by partitions based on their size using fix sized bins. Extends {@link
+ * SizeBasedFileRewritePlanner} with delete file number and delete ratio thresholds and job {@link
+ * RewriteDataFiles#REWRITE_JOB_ORDER} handling.
+ */
+public class BinPackRewriteFileGroupPlanner
Review Comment:
BinPackRewriteFilePlanner is probably more accurate
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2016134878
##########
core/src/main/java/org/apache/iceberg/actions/RewritePositionDeletesGroupPlanner.java:
##########
@@ -0,0 +1,221 @@
+/*
+ * 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.actions;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.MetadataTableType;
+import org.apache.iceberg.MetadataTableUtils;
+import org.apache.iceberg.Partitioning;
+import org.apache.iceberg.PositionDeletesScanTask;
+import org.apache.iceberg.PositionDeletesTable;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.actions.RewritePositionDeleteFiles.FileGroupInfo;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.PartitionUtil;
+import org.apache.iceberg.util.PropertyUtil;
+import org.apache.iceberg.util.StructLikeMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Groups specified delete files in the {@link Table} into {@link RewritePositionDeletesGroup}s.
+ * These will be grouped by partitions based on their size using fix sized bins. Extends the {@link
+ * SizeBasedFileRewritePlanner} with {@link RewritePositionDeleteFiles#REWRITE_JOB_ORDER} handling.
+ */
+public class RewritePositionDeletesGroupPlanner
+ extends SizeBasedFileRewritePlanner<
+ FileGroupInfo, PositionDeletesScanTask, DeleteFile, RewritePositionDeletesGroup> {
+ private static final Logger LOG =
+ LoggerFactory.getLogger(RewritePositionDeletesGroupPlanner.class);
+
+ private final Expression filter;
+ private final boolean caseSensitive;
+ private RewriteJobOrder rewriteJobOrder;
+
+ public RewritePositionDeletesGroupPlanner(Table table) {
+ this(table, Expressions.alwaysTrue(), false);
+ }
+
+ /**
+ * Creates the planner for the given table.
+ *
+ * @param table to plan for
+ * @param filter used to remove files from the plan
+ * @param caseSensitive property used for scanning
+ */
+ public RewritePositionDeletesGroupPlanner(Table table, Expression filter, boolean caseSensitive) {
+ super(table);
+ this.caseSensitive = caseSensitive;
+ this.filter = filter;
+ }
+
+ @Override
+ public Set<String> validOptions() {
+ return ImmutableSet.<String>builder()
+ .addAll(super.validOptions())
+ .add(RewritePositionDeleteFiles.REWRITE_JOB_ORDER)
+ .build();
+ }
+
+ @Override
+ public void init(Map<String, String> options) {
+ super.init(options);
+ this.rewriteJobOrder =
+ RewriteJobOrder.fromName(
+ PropertyUtil.propertyAsString(
+ options,
+ RewritePositionDeleteFiles.REWRITE_JOB_ORDER,
+ RewritePositionDeleteFiles.REWRITE_JOB_ORDER_DEFAULT));
+ }
+
+ @Override
+ public FileRewritePlan<
+ FileGroupInfo, PositionDeletesScanTask, DeleteFile, RewritePositionDeletesGroup>
+ plan() {
+ StructLikeMap<List<List<PositionDeletesScanTask>>> plan = planFileGroups();
+ RewriteExecutionContext ctx = new RewriteExecutionContext();
+ Stream<RewritePositionDeletesGroup> groups =
+ plan.entrySet().stream()
+ .filter(e -> !e.getValue().isEmpty())
+ .flatMap(
+ e -> {
+ StructLike partition = e.getKey();
+ List<List<PositionDeletesScanTask>> scanGroups = e.getValue();
+ return scanGroups.stream()
+ .map(
+ tasks -> {
+ long inputSize = inputSize(tasks);
+ return newRewriteGroup(
+ ctx,
+ partition,
+ tasks,
+ inputSplitSize(inputSize),
+ expectedOutputFiles(inputSize));
+ });
+ })
+ .sorted(RewritePositionDeletesGroup.comparator(rewriteJobOrder));
+ Map<StructLike, Integer> groupsInPartition = plan.transformValues(List::size);
+ int totalGroupCount = groupsInPartition.values().stream().reduce(Integer::sum).orElse(0);
+ return new FileRewritePlan<>(
+ CloseableIterable.of(groups.collect(Collectors.toList())),
+ totalGroupCount,
+ groupsInPartition);
+ }
+
+ @Override
+ protected Iterable<PositionDeletesScanTask> filterFiles(Iterable<PositionDeletesScanTask> tasks) {
+ return Iterables.filter(tasks, this::wronglySized);
+ }
+
+ @Override
+ protected Iterable<List<PositionDeletesScanTask>> filterFileGroups(
+ List<List<PositionDeletesScanTask>> groups) {
+ return Iterables.filter(
+ groups, group -> enoughInputFiles(group) || enoughContent(group) || tooMuchContent(group));
+ }
+
+ @Override
+ protected long defaultTargetFileSize() {
+ return PropertyUtil.propertyAsLong(
+ table().properties(),
+ TableProperties.DELETE_TARGET_FILE_SIZE_BYTES,
+ TableProperties.DELETE_TARGET_FILE_SIZE_BYTES_DEFAULT);
+ }
+
+ private StructLikeMap<List<List<PositionDeletesScanTask>>> planFileGroups() {
+ Table deletesTable =
+ MetadataTableUtils.createMetadataTableInstance(table(), MetadataTableType.POSITION_DELETES);
+ CloseableIterable<PositionDeletesScanTask> fileTasks = planFiles(deletesTable);
+
+ try {
+ Types.StructType partitionType = Partitioning.partitionType(deletesTable);
+ StructLikeMap<List<PositionDeletesScanTask>> fileTasksByPartition =
+ groupByPartition(partitionType, fileTasks);
+ return fileTasksByPartition.transformValues(
+ tasks -> ImmutableList.copyOf(planFileGroups(tasks)));
+ } finally {
+ try {
+ fileTasks.close();
+ } catch (IOException io) {
+ LOG.error("Cannot properly close file iterable while planning for rewrite", io);
+ }
+ }
+ }
+
+ private CloseableIterable<PositionDeletesScanTask> planFiles(Table deletesTable) {
+ PositionDeletesTable.PositionDeletesBatchScan scan =
+ (PositionDeletesTable.PositionDeletesBatchScan) deletesTable.newBatchScan();
+ return CloseableIterable.transform(
+ scan.baseTableFilter(filter).caseSensitive(caseSensitive).ignoreResiduals().planFiles(),
+ PositionDeletesScanTask.class::cast);
+ }
+
+ private StructLikeMap<List<PositionDeletesScanTask>> groupByPartition(
Review Comment:
I had the same feeling, and I tried it several times, but there are nuances which makes them different, and hard to generalize:
- The input tasks are different
- The partition handling is different (data file compaction handles partition evolution)
- The file filtering logic is different
- The group filterig logic is different
- Result groups are different
I see the partition handling as the main issue.
When compacting data files we have this additional logic:
```
// If a task uses an incompatible partition spec the data inside could contain values
// which belong to multiple partitions in the current spec. Treating all such files as
// un-partitioned and grouping them together helps to minimize new files made.
```
When compacting position delete files just uses `coercePartition` to update the task partition to the current one.
Based on these, I decided against the generalization.
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2016142918
##########
core/src/test/java/org/apache/iceberg/actions/TestBinPackRewriteFileGroupPlanner.java:
##########
@@ -0,0 +1,468 @@
+/*
+ * 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.actions;
+
+import static org.apache.iceberg.actions.BinPackRewriteFileGroupPlanner.MAX_FILE_SIZE_DEFAULT_RATIO;
+import static org.apache.iceberg.actions.RewriteDataFiles.REWRITE_JOB_ORDER;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.File;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.stream.Collectors;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.MockFileScanTask;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.TestBase;
+import org.apache.iceberg.TestTables;
+import org.apache.iceberg.actions.RewriteDataFiles.FileGroupInfo;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class TestBinPackRewriteFileGroupPlanner {
+ private static final Map<String, String> REWRITE_ALL =
+ ImmutableMap.of(BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true");
+
+ private static final DataFile FILE_1 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_2 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_3 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_4 = newDataFile("data_bucket=1", 11);
+ private static final DataFile FILE_5 = newDataFile("data_bucket=1", 11);
+ private static final DataFile FILE_6 = newDataFile("data_bucket=2", 50);
+
+ private static final Map<RewriteJobOrder, List<StructLike>> EXPECTED =
+ ImmutableMap.of(
+ RewriteJobOrder.FILES_DESC,
+ ImmutableList.of(FILE_1.partition(), FILE_4.partition(), FILE_6.partition()),
+ RewriteJobOrder.FILES_ASC,
+ ImmutableList.of(FILE_6.partition(), FILE_4.partition(), FILE_1.partition()),
+ RewriteJobOrder.BYTES_DESC,
+ ImmutableList.of(FILE_6.partition(), FILE_1.partition(), FILE_4.partition()),
+ RewriteJobOrder.BYTES_ASC,
+ ImmutableList.of(FILE_4.partition(), FILE_1.partition(), FILE_6.partition()));
+
+ @TempDir private File tableDir = null;
+ private TestTables.TestTable table = null;
+
+ @BeforeEach
+ public void setupTable() throws Exception {
+ this.table = TestTables.create(tableDir, "test", TestBase.SCHEMA, TestBase.SPEC, 3);
+ }
+
+ @AfterEach
+ public void cleanupTables() {
+ TestTables.clearTables();
+ }
+
+ @Test
+ void testPartitionedTable() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(ImmutableMap.of(BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups).hasSize(3);
+ assertThat(groups.stream().mapToInt(RewriteGroupBase::inputFileNum).sum()).isEqualTo(6);
+ }
+
+ @Test
+ void testUnpartitionedTable() {
+ table.updateSpec().removeField("data_bucket").commit();
+ table.refresh();
+ table
+ .newAppend()
+ .appendFile(newDataFile("", 10))
+ .appendFile(newDataFile("", 20))
+ .appendFile(newDataFile("", 30))
+ .commit();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.MIN_INPUT_FILES,
+ "1",
+ BinPackRewriteFileGroupPlanner.MIN_FILE_SIZE_BYTES,
+ "30"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.totalGroupCount()).isEqualTo(1);
+ assertThat(plan.groups().iterator().next().inputFileNum()).isEqualTo(2);
+ }
+
+ @Test
+ void testEmptyTable() {
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(REWRITE_ALL);
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(table.currentSnapshot()).as("Table must be empty").isNull();
+ assertThat(plan.totalGroupCount()).isZero();
+ }
+
+ @ParameterizedTest
+ @EnumSource(
+ value = RewriteJobOrder.class,
+ names = {"FILES_DESC", "FILES_ASC", "BYTES_DESC", "BYTES_ASC"})
+ void testJobOrder(RewriteJobOrder order) {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true", REWRITE_JOB_ORDER, order.name()));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups.stream().map(group -> group.info().partition()).collect(Collectors.toList()))
+ .isEqualTo(EXPECTED.get(order));
+ assertThat(plan.totalGroupCount()).isEqualTo(3);
+ EXPECTED.get(order).forEach(s -> assertThat(plan.groupsInPartition(s)).isEqualTo(1));
+ }
+
+ @Test
+ void testMaxGroupSize() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.MAX_FILE_GROUP_SIZE_BYTES,
+ "10"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.totalGroupCount()).isEqualTo(6);
+ assertThat(plan.groupsInPartition(FILE_1.partition())).isEqualTo(3);
+ assertThat(plan.groupsInPartition(FILE_4.partition())).isEqualTo(2);
+ assertThat(plan.groupsInPartition(FILE_6.partition())).isEqualTo(1);
+ }
+
+ @Test
+ void testFilter() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner =
+ new BinPackRewriteFileGroupPlanner(
+ table,
+ Expressions.or(
+ Expressions.equal(Expressions.bucket("data", 16), 0),
+ Expressions.equal(Expressions.bucket("data", 16), 2)));
+ planner.init(REWRITE_ALL);
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(plan.totalGroupCount()).isEqualTo(2);
+ assertThat(groups).hasSize(2);
+ assertThat(groups.stream().mapToLong(RewriteFileGroup::inputFileNum).sum()).isEqualTo(4);
+ }
+
+ @Test
+ void testMaxOutputFileSize() {
+ addFiles();
+ int targetFileSize = 10;
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.TARGET_FILE_SIZE_BYTES,
+ String.valueOf(targetFileSize)));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.groups().iterator().next().maxOutputFileSize())
+ .isGreaterThan(targetFileSize)
+ .isLessThan((long) (targetFileSize * MAX_FILE_SIZE_DEFAULT_RATIO));
+ }
+
+ @Test
+ void testExpectedOutputFiles() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.TARGET_FILE_SIZE_BYTES,
+ "21"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups).hasSize(3);
+ for (RewriteFileGroup group : groups) {
+ if (FILE_1.partition().equals(group.info().partition())) {
+ assertThat(group.expectedOutputFiles()).isEqualTo(2);
+ } else if (FILE_4.partition().equals(group.info().partition())) {
+ assertThat(group.expectedOutputFiles()).isEqualTo(1);
Review Comment:
This one is a bit tricky.
We allow file size 10% bigger than the target. See: `SizeBasedFileRewritePlanner.expectedOutputFiles` method
```
* [..] If the new average file size is no more than 10% greater than
* the target file size, then this method will round down when determining the number of output
* files. Otherwise, the remainder will be written into a separate file.
```
Added some comment to describe the tests case
##########
core/src/test/java/org/apache/iceberg/actions/TestBinPackRewriteFileGroupPlanner.java:
##########
@@ -0,0 +1,468 @@
+/*
+ * 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.actions;
+
+import static org.apache.iceberg.actions.BinPackRewriteFileGroupPlanner.MAX_FILE_SIZE_DEFAULT_RATIO;
+import static org.apache.iceberg.actions.RewriteDataFiles.REWRITE_JOB_ORDER;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.File;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.stream.Collectors;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.MockFileScanTask;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.TestBase;
+import org.apache.iceberg.TestTables;
+import org.apache.iceberg.actions.RewriteDataFiles.FileGroupInfo;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class TestBinPackRewriteFileGroupPlanner {
+ private static final Map<String, String> REWRITE_ALL =
+ ImmutableMap.of(BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true");
+
+ private static final DataFile FILE_1 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_2 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_3 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_4 = newDataFile("data_bucket=1", 11);
+ private static final DataFile FILE_5 = newDataFile("data_bucket=1", 11);
+ private static final DataFile FILE_6 = newDataFile("data_bucket=2", 50);
+
+ private static final Map<RewriteJobOrder, List<StructLike>> EXPECTED =
+ ImmutableMap.of(
+ RewriteJobOrder.FILES_DESC,
+ ImmutableList.of(FILE_1.partition(), FILE_4.partition(), FILE_6.partition()),
+ RewriteJobOrder.FILES_ASC,
+ ImmutableList.of(FILE_6.partition(), FILE_4.partition(), FILE_1.partition()),
+ RewriteJobOrder.BYTES_DESC,
+ ImmutableList.of(FILE_6.partition(), FILE_1.partition(), FILE_4.partition()),
+ RewriteJobOrder.BYTES_ASC,
+ ImmutableList.of(FILE_4.partition(), FILE_1.partition(), FILE_6.partition()));
+
+ @TempDir private File tableDir = null;
+ private TestTables.TestTable table = null;
+
+ @BeforeEach
+ public void setupTable() throws Exception {
+ this.table = TestTables.create(tableDir, "test", TestBase.SCHEMA, TestBase.SPEC, 3);
+ }
+
+ @AfterEach
+ public void cleanupTables() {
+ TestTables.clearTables();
+ }
+
+ @Test
+ void testPartitionedTable() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(ImmutableMap.of(BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups).hasSize(3);
+ assertThat(groups.stream().mapToInt(RewriteGroupBase::inputFileNum).sum()).isEqualTo(6);
+ }
+
+ @Test
+ void testUnpartitionedTable() {
+ table.updateSpec().removeField("data_bucket").commit();
+ table.refresh();
+ table
+ .newAppend()
+ .appendFile(newDataFile("", 10))
+ .appendFile(newDataFile("", 20))
+ .appendFile(newDataFile("", 30))
+ .commit();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.MIN_INPUT_FILES,
+ "1",
+ BinPackRewriteFileGroupPlanner.MIN_FILE_SIZE_BYTES,
+ "30"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.totalGroupCount()).isEqualTo(1);
+ assertThat(plan.groups().iterator().next().inputFileNum()).isEqualTo(2);
+ }
+
+ @Test
+ void testEmptyTable() {
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(REWRITE_ALL);
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(table.currentSnapshot()).as("Table must be empty").isNull();
+ assertThat(plan.totalGroupCount()).isZero();
+ }
+
+ @ParameterizedTest
+ @EnumSource(
+ value = RewriteJobOrder.class,
+ names = {"FILES_DESC", "FILES_ASC", "BYTES_DESC", "BYTES_ASC"})
+ void testJobOrder(RewriteJobOrder order) {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true", REWRITE_JOB_ORDER, order.name()));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups.stream().map(group -> group.info().partition()).collect(Collectors.toList()))
+ .isEqualTo(EXPECTED.get(order));
+ assertThat(plan.totalGroupCount()).isEqualTo(3);
+ EXPECTED.get(order).forEach(s -> assertThat(plan.groupsInPartition(s)).isEqualTo(1));
+ }
+
+ @Test
+ void testMaxGroupSize() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.MAX_FILE_GROUP_SIZE_BYTES,
+ "10"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.totalGroupCount()).isEqualTo(6);
+ assertThat(plan.groupsInPartition(FILE_1.partition())).isEqualTo(3);
+ assertThat(plan.groupsInPartition(FILE_4.partition())).isEqualTo(2);
+ assertThat(plan.groupsInPartition(FILE_6.partition())).isEqualTo(1);
+ }
+
+ @Test
+ void testFilter() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner =
+ new BinPackRewriteFileGroupPlanner(
+ table,
+ Expressions.or(
+ Expressions.equal(Expressions.bucket("data", 16), 0),
+ Expressions.equal(Expressions.bucket("data", 16), 2)));
+ planner.init(REWRITE_ALL);
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(plan.totalGroupCount()).isEqualTo(2);
+ assertThat(groups).hasSize(2);
+ assertThat(groups.stream().mapToLong(RewriteFileGroup::inputFileNum).sum()).isEqualTo(4);
+ }
+
+ @Test
+ void testMaxOutputFileSize() {
+ addFiles();
+ int targetFileSize = 10;
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.TARGET_FILE_SIZE_BYTES,
+ String.valueOf(targetFileSize)));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.groups().iterator().next().maxOutputFileSize())
+ .isGreaterThan(targetFileSize)
+ .isLessThan((long) (targetFileSize * MAX_FILE_SIZE_DEFAULT_RATIO));
+ }
+
+ @Test
+ void testExpectedOutputFiles() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.TARGET_FILE_SIZE_BYTES,
+ "21"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups).hasSize(3);
+ for (RewriteFileGroup group : groups) {
+ if (FILE_1.partition().equals(group.info().partition())) {
+ assertThat(group.expectedOutputFiles()).isEqualTo(2);
+ } else if (FILE_4.partition().equals(group.info().partition())) {
+ assertThat(group.expectedOutputFiles()).isEqualTo(1);
+ } else if (FILE_6.partition().equals(group.info().partition())) {
+ assertThat(group.expectedOutputFiles()).isEqualTo(3);
+ } else {
+ throw new IllegalStateException("Unexpected partition: " + group.info().partition());
+ }
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testOutputSpec(boolean specific) {
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2016152148
##########
core/src/test/java/org/apache/iceberg/actions/TestBinPackRewriteFileGroupPlanner.java:
##########
@@ -0,0 +1,468 @@
+/*
+ * 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.actions;
+
+import static org.apache.iceberg.actions.BinPackRewriteFileGroupPlanner.MAX_FILE_SIZE_DEFAULT_RATIO;
+import static org.apache.iceberg.actions.RewriteDataFiles.REWRITE_JOB_ORDER;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.File;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.stream.Collectors;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.MockFileScanTask;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.TestBase;
+import org.apache.iceberg.TestTables;
+import org.apache.iceberg.actions.RewriteDataFiles.FileGroupInfo;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class TestBinPackRewriteFileGroupPlanner {
+ private static final Map<String, String> REWRITE_ALL =
+ ImmutableMap.of(BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true");
+
+ private static final DataFile FILE_1 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_2 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_3 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_4 = newDataFile("data_bucket=1", 11);
+ private static final DataFile FILE_5 = newDataFile("data_bucket=1", 11);
+ private static final DataFile FILE_6 = newDataFile("data_bucket=2", 50);
+
+ private static final Map<RewriteJobOrder, List<StructLike>> EXPECTED =
+ ImmutableMap.of(
+ RewriteJobOrder.FILES_DESC,
+ ImmutableList.of(FILE_1.partition(), FILE_4.partition(), FILE_6.partition()),
+ RewriteJobOrder.FILES_ASC,
+ ImmutableList.of(FILE_6.partition(), FILE_4.partition(), FILE_1.partition()),
+ RewriteJobOrder.BYTES_DESC,
+ ImmutableList.of(FILE_6.partition(), FILE_1.partition(), FILE_4.partition()),
+ RewriteJobOrder.BYTES_ASC,
+ ImmutableList.of(FILE_4.partition(), FILE_1.partition(), FILE_6.partition()));
+
+ @TempDir private File tableDir = null;
+ private TestTables.TestTable table = null;
+
+ @BeforeEach
+ public void setupTable() throws Exception {
+ this.table = TestTables.create(tableDir, "test", TestBase.SCHEMA, TestBase.SPEC, 3);
+ }
+
+ @AfterEach
+ public void cleanupTables() {
+ TestTables.clearTables();
+ }
+
+ @Test
+ void testPartitionedTable() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(ImmutableMap.of(BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups).hasSize(3);
+ assertThat(groups.stream().mapToInt(RewriteGroupBase::inputFileNum).sum()).isEqualTo(6);
+ }
+
+ @Test
+ void testUnpartitionedTable() {
+ table.updateSpec().removeField("data_bucket").commit();
+ table.refresh();
+ table
+ .newAppend()
+ .appendFile(newDataFile("", 10))
+ .appendFile(newDataFile("", 20))
+ .appendFile(newDataFile("", 30))
+ .commit();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.MIN_INPUT_FILES,
+ "1",
+ BinPackRewriteFileGroupPlanner.MIN_FILE_SIZE_BYTES,
+ "30"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.totalGroupCount()).isEqualTo(1);
+ assertThat(plan.groups().iterator().next().inputFileNum()).isEqualTo(2);
+ }
+
+ @Test
+ void testEmptyTable() {
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(REWRITE_ALL);
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(table.currentSnapshot()).as("Table must be empty").isNull();
+ assertThat(plan.totalGroupCount()).isZero();
+ }
+
+ @ParameterizedTest
+ @EnumSource(
+ value = RewriteJobOrder.class,
+ names = {"FILES_DESC", "FILES_ASC", "BYTES_DESC", "BYTES_ASC"})
+ void testJobOrder(RewriteJobOrder order) {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true", REWRITE_JOB_ORDER, order.name()));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups.stream().map(group -> group.info().partition()).collect(Collectors.toList()))
+ .isEqualTo(EXPECTED.get(order));
+ assertThat(plan.totalGroupCount()).isEqualTo(3);
+ EXPECTED.get(order).forEach(s -> assertThat(plan.groupsInPartition(s)).isEqualTo(1));
+ }
+
+ @Test
+ void testMaxGroupSize() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.MAX_FILE_GROUP_SIZE_BYTES,
+ "10"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.totalGroupCount()).isEqualTo(6);
+ assertThat(plan.groupsInPartition(FILE_1.partition())).isEqualTo(3);
+ assertThat(plan.groupsInPartition(FILE_4.partition())).isEqualTo(2);
+ assertThat(plan.groupsInPartition(FILE_6.partition())).isEqualTo(1);
+ }
+
+ @Test
+ void testFilter() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner =
+ new BinPackRewriteFileGroupPlanner(
+ table,
+ Expressions.or(
+ Expressions.equal(Expressions.bucket("data", 16), 0),
+ Expressions.equal(Expressions.bucket("data", 16), 2)));
+ planner.init(REWRITE_ALL);
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(plan.totalGroupCount()).isEqualTo(2);
+ assertThat(groups).hasSize(2);
+ assertThat(groups.stream().mapToLong(RewriteFileGroup::inputFileNum).sum()).isEqualTo(4);
+ }
+
+ @Test
+ void testMaxOutputFileSize() {
+ addFiles();
+ int targetFileSize = 10;
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.TARGET_FILE_SIZE_BYTES,
+ String.valueOf(targetFileSize)));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.groups().iterator().next().maxOutputFileSize())
+ .isGreaterThan(targetFileSize)
+ .isLessThan((long) (targetFileSize * MAX_FILE_SIZE_DEFAULT_RATIO));
+ }
+
+ @Test
+ void testExpectedOutputFiles() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.TARGET_FILE_SIZE_BYTES,
+ "21"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups).hasSize(3);
+ for (RewriteFileGroup group : groups) {
+ if (FILE_1.partition().equals(group.info().partition())) {
+ assertThat(group.expectedOutputFiles()).isEqualTo(2);
+ } else if (FILE_4.partition().equals(group.info().partition())) {
+ assertThat(group.expectedOutputFiles()).isEqualTo(1);
+ } else if (FILE_6.partition().equals(group.info().partition())) {
+ assertThat(group.expectedOutputFiles()).isEqualTo(3);
+ } else {
+ throw new IllegalStateException("Unexpected partition: " + group.info().partition());
+ }
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testOutputSpec(boolean specific) {
+ addFiles();
+ int oldSpecId = table.spec().specId();
+ table.updateSpec().removeField("data_bucket").commit();
+ table.newAppend().appendFile(newDataFile("", 10)).commit();
+ table.refresh();
+ int newSpecId = table.spec().specId();
+
+ Map<String, String> options = Maps.newHashMap(REWRITE_ALL);
+ if (specific) {
+ options.put(RewriteDataFiles.OUTPUT_SPEC_ID, String.valueOf(oldSpecId));
+ }
+
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(options);
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.groups().iterator().next().outputSpecId())
+ .isEqualTo(specific ? oldSpecId : newSpecId);
+ }
+
+ @Test
+ void testValidOptions() {
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+
+ assertThat(planner.validOptions())
+ .as("Planner must report all supported options")
+ .isEqualTo(
+ ImmutableSet.of(
+ BinPackRewriteFileGroupPlanner.TARGET_FILE_SIZE_BYTES,
+ BinPackRewriteFileGroupPlanner.MIN_FILE_SIZE_BYTES,
+ BinPackRewriteFileGroupPlanner.MAX_FILE_SIZE_BYTES,
+ BinPackRewriteFileGroupPlanner.MIN_INPUT_FILES,
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ BinPackRewriteFileGroupPlanner.MAX_FILE_GROUP_SIZE_BYTES,
+ BinPackRewriteFileGroupPlanner.DELETE_FILE_THRESHOLD,
+ BinPackRewriteFileGroupPlanner.DELETE_RATIO_THRESHOLD,
+ RewriteDataFiles.REWRITE_JOB_ORDER));
+ }
+
+ @Test
+ void testInvalidOption() {
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+
+ Map<String, String> invalidRewriteJobOrderOptions =
+ ImmutableMap.of(RewriteDataFiles.REWRITE_JOB_ORDER, "foo");
+ assertThatThrownBy(() -> planner.init(invalidRewriteJobOrderOptions))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid rewrite job order name: foo");
+
+ Map<String, String> invalidOutputSpecIdOptions =
+ ImmutableMap.of(RewriteDataFiles.OUTPUT_SPEC_ID, String.valueOf(1234));
+ assertThatThrownBy(() -> planner.init(invalidOutputSpecIdOptions))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage(
+ "Cannot use output spec id 1234 because the table does not contain a reference to this spec-id.");
+
+ Map<String, String> invalidDeleteFileThresholdOptions =
+ ImmutableMap.of(BinPackRewriteFileGroupPlanner.DELETE_FILE_THRESHOLD, "-1");
+ assertThatThrownBy(() -> planner.init(invalidDeleteFileThresholdOptions))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("'delete-file-threshold' is set to -1 but must be >= 0");
+
+ Map<String, String> negativeDeleteRatioThresholdOptions =
+ ImmutableMap.of(BinPackRewriteFileGroupPlanner.DELETE_RATIO_THRESHOLD, "-1");
+ assertThatThrownBy(() -> planner.init(negativeDeleteRatioThresholdOptions))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("'delete-ratio-threshold' is set to -1.0 but must be > 0");
+
+ Map<String, String> invalidDeleteRatioThresholdOptions =
+ ImmutableMap.of(BinPackRewriteFileGroupPlanner.DELETE_RATIO_THRESHOLD, "127");
+ assertThatThrownBy(() -> planner.init(invalidDeleteRatioThresholdOptions))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("'delete-ratio-threshold' is set to 127.0 but must be <= 1");
+ }
+
+ @Test
+ void testSizeFiltering() {
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ FileScanTask tooSmallTask = new MockFileScanTask(100L);
+ FileScanTask optimal = new MockFileScanTask(450);
+ FileScanTask tooBigTask = new MockFileScanTask(1000L);
+ List<FileScanTask> tasks = ImmutableList.of(tooSmallTask, optimal, tooBigTask);
+
+ Map<String, String> options =
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.MIN_FILE_SIZE_BYTES, "250",
+ BinPackRewriteFileGroupPlanner.TARGET_FILE_SIZE_BYTES, "500",
+ BinPackRewriteFileGroupPlanner.MAX_FILE_SIZE_BYTES, "750");
+ planner.init(options);
+
+ Iterable<List<FileScanTask>> groups = planner.planFileGroups(tasks);
+ assertThat(groups).as("Must have 1 group").hasSize(1);
+ List<FileScanTask> group = Iterables.getOnlyElement(groups);
+ assertThat(group).as("Must rewrite 2 files").hasSize(2);
+ assertThat(group).containsExactlyInAnyOrder(tooSmallTask, tooBigTask);
+ }
+
+ @Test
+ void testGroupWithEnoughFiles() {
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ List<FileScanTask> tasks =
+ ImmutableList.of(
+ new MockFileScanTask(100L),
+ new MockFileScanTask(100L),
+ new MockFileScanTask(100L),
+ new MockFileScanTask(100L));
+
+ Map<String, String> options =
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.MIN_INPUT_FILES, "3",
+ BinPackRewriteFileGroupPlanner.MIN_FILE_SIZE_BYTES, "150",
+ BinPackRewriteFileGroupPlanner.TARGET_FILE_SIZE_BYTES, "1000",
+ BinPackRewriteFileGroupPlanner.MAX_FILE_SIZE_BYTES, "5000");
+ planner.init(options);
+
+ Iterable<List<FileScanTask>> groups = planner.planFileGroups(tasks);
+ assertThat(groups).as("Must have 1 group").hasSize(1);
+ List<FileScanTask> group = Iterables.getOnlyElement(groups);
+ assertThat(group).as("Must rewrite 4 files").hasSize(4);
+ }
+
+ @Test
+ void testGroupWithEnoughData() {
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ List<FileScanTask> tasks =
+ ImmutableList.of(
+ new MockFileScanTask(100L), new MockFileScanTask(100L), new MockFileScanTask(100L));
+
+ Map<String, String> options =
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.MIN_FILE_SIZE_BYTES, "200",
+ BinPackRewriteFileGroupPlanner.TARGET_FILE_SIZE_BYTES, "250",
Review Comment:
The number of planned groups depends on the `MAX_FILE_SIZE_BYTES`. The number of the files we would like to rewrite depends on the `MIN_FILE_SIZE_BYTES` and `TARGET_FILE_SIZE_BYTES`.
Since all of the files are 100 bytes long in the test, all of them will be target for rewrite. This is true for both `200` or `300`. The only difference is the resulting `expectedOutputFiles` which should be 2 with `200` and 1 with `300`.
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2014688814
##########
core/src/main/java/org/apache/iceberg/actions/RewriteFileGroupPlanner.java:
##########
@@ -0,0 +1,310 @@
+/*
+ * 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.actions;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.actions.RewriteDataFiles.FileGroupInfo;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.ContentFileUtil;
+import org.apache.iceberg.util.PropertyUtil;
+import org.apache.iceberg.util.StructLikeMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Groups specified data files in the {@link Table} into {@link RewriteFileGroup}s. The files are
+ * grouped by partitions based on their size using fix sized bins. Extends {@link
+ * SizeBasedFileRewritePlanner} with delete file number and delete ratio thresholds and job {@link
+ * RewriteDataFiles#REWRITE_JOB_ORDER} handling.
+ */
+public class RewriteFileGroupPlanner
+ extends SizeBasedFileRewritePlanner<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> {
+ /**
+ * The minimum number of deletes that needs to be associated with a data file for it to be
+ * considered for rewriting. If a data file has this number of deletes or more, it will be
+ * rewritten regardless of its file size determined by {@link #MIN_FILE_SIZE_BYTES} and {@link
+ * #MAX_FILE_SIZE_BYTES}. If a file group contains a file that satisfies this condition, the file
+ * group will be rewritten regardless of the number of files in the file group determined by
+ * {@link #MIN_INPUT_FILES}.
+ *
+ * <p>Defaults to Integer.MAX_VALUE, which means this feature is not enabled by default.
+ */
+ public static final String DELETE_FILE_THRESHOLD = "delete-file-threshold";
+
+ public static final int DELETE_FILE_THRESHOLD_DEFAULT = Integer.MAX_VALUE;
+
+ /**
+ * The minimum deletion ratio that needs to be associated with a data file for it to be considered
Review Comment:
a few comments
* `percentage` is probably not the most accurate here, which would imply the default value as 30.0 (percent)
* `greater than or equal to`
```
/**
* The ratio of the deleted rows in a data file for it to be considered for rewriting. If the
* deletion ratio of a data file is greater than or equal to this value, it will be rewritten
* regardless of its file size determined by {@link #MIN_FILE_SIZE_BYTES} and {@link
* #MAX_FILE_SIZE_BYTES}. If a file group contains a file that satisfies this condition, the file
* group will be rewritten regardless of the number of files in the file group determined by
* {@link #MIN_INPUT_FILES}.
*
* <p>Defaults to 0.3, which means that if the number of deleted records in a file reaches or
* exceeds 30%, it may trigger the rewriting operation.
*/
```
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2014994662
##########
core/src/main/java/org/apache/iceberg/actions/SizeBasedDataRewriter.java:
##########
@@ -49,15 +56,15 @@ public abstract class SizeBasedDataRewriter extends SizeBasedFileRewriter<FileSc
public static final int DELETE_FILE_THRESHOLD_DEFAULT = Integer.MAX_VALUE;
/**
- * The minimum deletion ratio that needs to be associated with a data file for it to be considered
- * for rewriting. If the deletion ratio of a data file is greater than or equal to this value, it
- * will be rewritten regardless of its file size determined by {@link #MIN_FILE_SIZE_BYTES} and
- * {@link #MAX_FILE_SIZE_BYTES}. If a file group contains a file that satisfies this condition,
- * the file group will be rewritten regardless of the number of files in the file group determined
- * by {@link #MIN_INPUT_FILES}.
+ * The percentage of the deleted records in a data file that enforces a data file rewrite. If the
+ * deletion ratio of a data file is greater or equal than this value, it will be rewritten
+ * regardless of its file size determined by {@link #MIN_FILE_SIZE_BYTES} and {@link
+ * #MAX_FILE_SIZE_BYTES}. If a file group contains a file that satisfies this condition, the file
+ * group will be rewritten regardless of the number of files in the file group determined by
+ * {@link #MIN_INPUT_FILES}.
*
- * <p>Defaults to 0.3, which means that if the deletion ratio of a file reaches or exceeds 30%, it
- * may trigger the rewriting operation.
+ * <p>Defaults to 0.3, which means that if the number of deleted records in a file reaches or
+ * exceeds 30%, it may trigger the rewriting operation.
Review Comment:
```suggestion
* exceeds 30%, it will trigger the rewriting operation.
```
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2016082527
##########
core/src/main/java/org/apache/iceberg/actions/RewritePositionDeletesGroupPlanner.java:
##########
@@ -0,0 +1,221 @@
+/*
+ * 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.actions;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.MetadataTableType;
+import org.apache.iceberg.MetadataTableUtils;
+import org.apache.iceberg.Partitioning;
+import org.apache.iceberg.PositionDeletesScanTask;
+import org.apache.iceberg.PositionDeletesTable;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.actions.RewritePositionDeleteFiles.FileGroupInfo;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.PartitionUtil;
+import org.apache.iceberg.util.PropertyUtil;
+import org.apache.iceberg.util.StructLikeMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Groups specified delete files in the {@link Table} into {@link RewritePositionDeletesGroup}s.
+ * These will be grouped by partitions based on their size using fix sized bins. Extends the {@link
+ * SizeBasedFileRewritePlanner} with {@link RewritePositionDeleteFiles#REWRITE_JOB_ORDER} handling.
+ */
+public class RewritePositionDeletesGroupPlanner
Review Comment:
Renamed to `BinPackRewritePositionDeletePlanner`
##########
core/src/main/java/org/apache/iceberg/actions/SizeBasedFileRewritePlanner.java:
##########
@@ -0,0 +1,361 @@
+/*
+ * 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.actions;
+
+import java.math.RoundingMode;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.ContentScanTask;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.math.LongMath;
+import org.apache.iceberg.util.BinPacking;
+import org.apache.iceberg.util.PropertyUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A file rewrite planner that determines which files to rewrite based on their size.
+ *
+ * <p>If files are smaller than the {@link #MIN_FILE_SIZE_BYTES} threshold or larger than the {@link
+ * #MAX_FILE_SIZE_BYTES} threshold, they are considered targets for being rewritten.
+ *
+ * <p>Once selected, files are grouped based on the {@link BinPacking bin-packing algorithm} into
+ * groups of no more than {@link #MAX_FILE_GROUP_SIZE_BYTES}. Groups will be actually rewritten if
+ * they contain more than {@link #MIN_INPUT_FILES} or if they would produce at least one file of
+ * {@link #TARGET_FILE_SIZE_BYTES}.
+ *
+ * <p>Note that implementations may add extra conditions for selecting files or filtering groups.
+ */
+public abstract class SizeBasedFileRewritePlanner<
+ I,
+ T extends ContentScanTask<F>,
+ F extends ContentFile<F>,
+ G extends RewriteGroupBase<I, T, F>>
+ implements FileRewritePlanner<I, T, F, G> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SizeBasedFileRewritePlanner.class);
+
+ /** The target output file size that this file rewriter will attempt to generate. */
+ public static final String TARGET_FILE_SIZE_BYTES = "target-file-size-bytes";
+
+ /**
+ * Controls which files will be considered for rewriting. Files with sizes under this threshold
+ * will be considered for rewriting regardless of any other criteria.
+ *
+ * <p>Defaults to 75% of the target file size.
+ */
+ public static final String MIN_FILE_SIZE_BYTES = "min-file-size-bytes";
+
+ public static final double MIN_FILE_SIZE_DEFAULT_RATIO = 0.75;
+
+ /**
+ * Controls which files will be considered for rewriting. Files with sizes above this threshold
+ * will be considered for rewriting regardless of any other criteria.
+ *
+ * <p>Defaults to 180% of the target file size.
+ */
+ public static final String MAX_FILE_SIZE_BYTES = "max-file-size-bytes";
+
+ public static final double MAX_FILE_SIZE_DEFAULT_RATIO = 1.80;
+
+ /**
+ * Any file group exceeding this number of files will be rewritten regardless of other criteria.
+ * This config ensures file groups that contain many files are compacted even if the total size of
+ * that group is less than the target file size. This can also be thought of as the maximum number
+ * of wrongly sized files that could remain in a partition after rewriting.
+ */
+ public static final String MIN_INPUT_FILES = "min-input-files";
+
+ public static final int MIN_INPUT_FILES_DEFAULT = 5;
+
+ /** Overrides other options and forces rewriting of all provided files. */
+ public static final String REWRITE_ALL = "rewrite-all";
+
+ public static final boolean REWRITE_ALL_DEFAULT = false;
+
+ /**
+ * This option controls the largest amount of data that should be rewritten in a single file
+ * group. It helps with breaking down the rewriting of very large partitions which may not be
+ * rewritable otherwise due to the resource constraints of the cluster. For example, a sort-based
+ * rewrite may not scale to TB-sized partitions, and those partitions need to be worked on in
+ * small subsections to avoid exhaustion of resources.
+ */
+ public static final String MAX_FILE_GROUP_SIZE_BYTES = "max-file-group-size-bytes";
+
+ public static final long MAX_FILE_GROUP_SIZE_BYTES_DEFAULT = 100L * 1024 * 1024 * 1024; // 100 GB
+
+ private static final long SPLIT_OVERHEAD = 5L * 1024;
+
+ private final Table table;
+ private long targetFileSize;
+ private long minFileSize;
+ private long maxFileSize;
+ private int minInputFiles;
+ private boolean rewriteAll;
+ private long maxGroupSize;
+ private int outputSpecId;
+
+ protected SizeBasedFileRewritePlanner(Table table) {
+ this.table = table;
+ }
+
+ /** Expected target file size before configuration. */
+ protected abstract long defaultTargetFileSize();
+
+ /** Additional filter for tasks before grouping. */
+ protected abstract Iterable<T> filterFiles(Iterable<T> tasks);
+
+ /** Additional filter for groups. */
+ protected abstract Iterable<List<T>> filterFileGroups(List<List<T>> groups);
+
+ @Override
+ public Set<String> validOptions() {
+ return ImmutableSet.of(
+ TARGET_FILE_SIZE_BYTES,
+ MIN_FILE_SIZE_BYTES,
+ MAX_FILE_SIZE_BYTES,
+ MIN_INPUT_FILES,
+ REWRITE_ALL,
+ MAX_FILE_GROUP_SIZE_BYTES);
+ }
+
+ @Override
+ public void init(Map<String, String> options) {
+ Map<String, Long> sizeThresholds = sizeThresholds(options);
+ this.targetFileSize = sizeThresholds.get(TARGET_FILE_SIZE_BYTES);
+ this.minFileSize = sizeThresholds.get(MIN_FILE_SIZE_BYTES);
+ this.maxFileSize = sizeThresholds.get(MAX_FILE_SIZE_BYTES);
+ this.minInputFiles = minInputFiles(options);
+ this.rewriteAll = rewriteAll(options);
+ this.maxGroupSize = maxGroupSize(options);
+ this.outputSpecId = outputSpecId(options);
+
+ if (rewriteAll) {
+ LOG.info("Configured to rewrite all provided files in table {}", table.name());
+ }
+ }
+
+ protected Table table() {
+ return table;
+ }
+
+ protected boolean wronglySized(T task) {
Review Comment:
Renamed
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2016139881
##########
core/src/test/java/org/apache/iceberg/actions/TestBinPackRewriteFileGroupPlanner.java:
##########
@@ -0,0 +1,468 @@
+/*
+ * 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.actions;
+
+import static org.apache.iceberg.actions.BinPackRewriteFileGroupPlanner.MAX_FILE_SIZE_DEFAULT_RATIO;
+import static org.apache.iceberg.actions.RewriteDataFiles.REWRITE_JOB_ORDER;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.File;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.stream.Collectors;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.MockFileScanTask;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.TestBase;
+import org.apache.iceberg.TestTables;
+import org.apache.iceberg.actions.RewriteDataFiles.FileGroupInfo;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class TestBinPackRewriteFileGroupPlanner {
+ private static final Map<String, String> REWRITE_ALL =
+ ImmutableMap.of(BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true");
+
+ private static final DataFile FILE_1 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_2 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_3 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_4 = newDataFile("data_bucket=1", 11);
+ private static final DataFile FILE_5 = newDataFile("data_bucket=1", 11);
+ private static final DataFile FILE_6 = newDataFile("data_bucket=2", 50);
+
+ private static final Map<RewriteJobOrder, List<StructLike>> EXPECTED =
+ ImmutableMap.of(
+ RewriteJobOrder.FILES_DESC,
+ ImmutableList.of(FILE_1.partition(), FILE_4.partition(), FILE_6.partition()),
+ RewriteJobOrder.FILES_ASC,
+ ImmutableList.of(FILE_6.partition(), FILE_4.partition(), FILE_1.partition()),
+ RewriteJobOrder.BYTES_DESC,
+ ImmutableList.of(FILE_6.partition(), FILE_1.partition(), FILE_4.partition()),
+ RewriteJobOrder.BYTES_ASC,
+ ImmutableList.of(FILE_4.partition(), FILE_1.partition(), FILE_6.partition()));
+
+ @TempDir private File tableDir = null;
+ private TestTables.TestTable table = null;
+
+ @BeforeEach
+ public void setupTable() throws Exception {
+ this.table = TestTables.create(tableDir, "test", TestBase.SCHEMA, TestBase.SPEC, 3);
+ }
+
+ @AfterEach
+ public void cleanupTables() {
+ TestTables.clearTables();
+ }
+
+ @Test
+ void testPartitionedTable() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(ImmutableMap.of(BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups).hasSize(3);
+ assertThat(groups.stream().mapToInt(RewriteGroupBase::inputFileNum).sum()).isEqualTo(6);
+ }
+
+ @Test
+ void testUnpartitionedTable() {
+ table.updateSpec().removeField("data_bucket").commit();
+ table.refresh();
+ table
+ .newAppend()
+ .appendFile(newDataFile("", 10))
+ .appendFile(newDataFile("", 20))
+ .appendFile(newDataFile("", 30))
+ .commit();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.MIN_INPUT_FILES,
+ "1",
+ BinPackRewriteFileGroupPlanner.MIN_FILE_SIZE_BYTES,
+ "30"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.totalGroupCount()).isEqualTo(1);
+ assertThat(plan.groups().iterator().next().inputFileNum()).isEqualTo(2);
+ }
+
+ @Test
+ void testEmptyTable() {
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(REWRITE_ALL);
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(table.currentSnapshot()).as("Table must be empty").isNull();
+ assertThat(plan.totalGroupCount()).isZero();
+ }
+
+ @ParameterizedTest
+ @EnumSource(
+ value = RewriteJobOrder.class,
+ names = {"FILES_DESC", "FILES_ASC", "BYTES_DESC", "BYTES_ASC"})
+ void testJobOrder(RewriteJobOrder order) {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true", REWRITE_JOB_ORDER, order.name()));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups.stream().map(group -> group.info().partition()).collect(Collectors.toList()))
+ .isEqualTo(EXPECTED.get(order));
+ assertThat(plan.totalGroupCount()).isEqualTo(3);
+ EXPECTED.get(order).forEach(s -> assertThat(plan.groupsInPartition(s)).isEqualTo(1));
+ }
+
+ @Test
+ void testMaxGroupSize() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.MAX_FILE_GROUP_SIZE_BYTES,
+ "10"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.totalGroupCount()).isEqualTo(6);
+ assertThat(plan.groupsInPartition(FILE_1.partition())).isEqualTo(3);
+ assertThat(plan.groupsInPartition(FILE_4.partition())).isEqualTo(2);
+ assertThat(plan.groupsInPartition(FILE_6.partition())).isEqualTo(1);
+ }
+
+ @Test
+ void testFilter() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner =
+ new BinPackRewriteFileGroupPlanner(
+ table,
+ Expressions.or(
+ Expressions.equal(Expressions.bucket("data", 16), 0),
+ Expressions.equal(Expressions.bucket("data", 16), 2)));
+ planner.init(REWRITE_ALL);
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(plan.totalGroupCount()).isEqualTo(2);
+ assertThat(groups).hasSize(2);
+ assertThat(groups.stream().mapToLong(RewriteFileGroup::inputFileNum).sum()).isEqualTo(4);
+ }
+
+ @Test
+ void testMaxOutputFileSize() {
+ addFiles();
+ int targetFileSize = 10;
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.TARGET_FILE_SIZE_BYTES,
+ String.valueOf(targetFileSize)));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.groups().iterator().next().maxOutputFileSize())
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2017027965
##########
core/src/test/java/org/apache/iceberg/actions/TestBinPackRewriteFileGroupPlanner.java:
##########
@@ -0,0 +1,468 @@
+/*
+ * 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.actions;
+
+import static org.apache.iceberg.actions.BinPackRewriteFileGroupPlanner.MAX_FILE_SIZE_DEFAULT_RATIO;
+import static org.apache.iceberg.actions.RewriteDataFiles.REWRITE_JOB_ORDER;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.File;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.stream.Collectors;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.MockFileScanTask;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.TestBase;
+import org.apache.iceberg.TestTables;
+import org.apache.iceberg.actions.RewriteDataFiles.FileGroupInfo;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class TestBinPackRewriteFileGroupPlanner {
+ private static final Map<String, String> REWRITE_ALL =
+ ImmutableMap.of(BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true");
+
+ private static final DataFile FILE_1 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_2 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_3 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_4 = newDataFile("data_bucket=1", 11);
+ private static final DataFile FILE_5 = newDataFile("data_bucket=1", 11);
+ private static final DataFile FILE_6 = newDataFile("data_bucket=2", 50);
+
+ private static final Map<RewriteJobOrder, List<StructLike>> EXPECTED =
+ ImmutableMap.of(
+ RewriteJobOrder.FILES_DESC,
+ ImmutableList.of(FILE_1.partition(), FILE_4.partition(), FILE_6.partition()),
+ RewriteJobOrder.FILES_ASC,
+ ImmutableList.of(FILE_6.partition(), FILE_4.partition(), FILE_1.partition()),
+ RewriteJobOrder.BYTES_DESC,
+ ImmutableList.of(FILE_6.partition(), FILE_1.partition(), FILE_4.partition()),
+ RewriteJobOrder.BYTES_ASC,
+ ImmutableList.of(FILE_4.partition(), FILE_1.partition(), FILE_6.partition()));
+
+ @TempDir private File tableDir = null;
+ private TestTables.TestTable table = null;
+
+ @BeforeEach
+ public void setupTable() throws Exception {
+ this.table = TestTables.create(tableDir, "test", TestBase.SCHEMA, TestBase.SPEC, 3);
+ }
+
+ @AfterEach
+ public void cleanupTables() {
+ TestTables.clearTables();
+ }
+
+ @Test
+ void testPartitionedTable() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(ImmutableMap.of(BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups).hasSize(3);
+ assertThat(groups.stream().mapToInt(RewriteGroupBase::inputFileNum).sum()).isEqualTo(6);
+ }
+
+ @Test
+ void testUnpartitionedTable() {
+ table.updateSpec().removeField("data_bucket").commit();
+ table.refresh();
+ table
+ .newAppend()
+ .appendFile(newDataFile("", 10))
+ .appendFile(newDataFile("", 20))
+ .appendFile(newDataFile("", 30))
+ .commit();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.MIN_INPUT_FILES,
+ "1",
+ BinPackRewriteFileGroupPlanner.MIN_FILE_SIZE_BYTES,
+ "30"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.totalGroupCount()).isEqualTo(1);
+ assertThat(plan.groups().iterator().next().inputFileNum()).isEqualTo(2);
+ }
+
+ @Test
+ void testEmptyTable() {
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(REWRITE_ALL);
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(table.currentSnapshot()).as("Table must be empty").isNull();
+ assertThat(plan.totalGroupCount()).isZero();
+ }
+
+ @ParameterizedTest
+ @EnumSource(
+ value = RewriteJobOrder.class,
+ names = {"FILES_DESC", "FILES_ASC", "BYTES_DESC", "BYTES_ASC"})
+ void testJobOrder(RewriteJobOrder order) {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true", REWRITE_JOB_ORDER, order.name()));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups.stream().map(group -> group.info().partition()).collect(Collectors.toList()))
+ .isEqualTo(EXPECTED.get(order));
+ assertThat(plan.totalGroupCount()).isEqualTo(3);
+ EXPECTED.get(order).forEach(s -> assertThat(plan.groupsInPartition(s)).isEqualTo(1));
+ }
+
+ @Test
+ void testMaxGroupSize() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.MAX_FILE_GROUP_SIZE_BYTES,
+ "10"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.totalGroupCount()).isEqualTo(6);
+ assertThat(plan.groupsInPartition(FILE_1.partition())).isEqualTo(3);
+ assertThat(plan.groupsInPartition(FILE_4.partition())).isEqualTo(2);
+ assertThat(plan.groupsInPartition(FILE_6.partition())).isEqualTo(1);
+ }
+
+ @Test
+ void testFilter() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner =
+ new BinPackRewriteFileGroupPlanner(
+ table,
+ Expressions.or(
+ Expressions.equal(Expressions.bucket("data", 16), 0),
+ Expressions.equal(Expressions.bucket("data", 16), 2)));
+ planner.init(REWRITE_ALL);
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(plan.totalGroupCount()).isEqualTo(2);
+ assertThat(groups).hasSize(2);
+ assertThat(groups.stream().mapToLong(RewriteFileGroup::inputFileNum).sum()).isEqualTo(4);
+ }
+
+ @Test
+ void testMaxOutputFileSize() {
+ addFiles();
+ int targetFileSize = 10;
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.TARGET_FILE_SIZE_BYTES,
+ String.valueOf(targetFileSize)));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.groups().iterator().next().maxOutputFileSize())
+ .isGreaterThan(targetFileSize)
+ .isLessThan((long) (targetFileSize * MAX_FILE_SIZE_DEFAULT_RATIO));
+ }
+
+ @Test
+ void testExpectedOutputFiles() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.TARGET_FILE_SIZE_BYTES,
+ "21"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups).hasSize(3);
+ for (RewriteFileGroup group : groups) {
+ if (FILE_1.partition().equals(group.info().partition())) {
+ assertThat(group.expectedOutputFiles()).isEqualTo(2);
+ } else if (FILE_4.partition().equals(group.info().partition())) {
+ assertThat(group.expectedOutputFiles()).isEqualTo(1);
Review Comment:
yes, that is the scenario I was asking. thx for confirming and clarifying in the code
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2013691646
##########
core/src/main/java/org/apache/iceberg/actions/RewriteFileGroup.java:
##########
@@ -48,14 +51,14 @@ public RewriteFileGroup(
List<FileScanTask> fileScanTasks,
int outputSpecId,
long writeMaxFileSize,
- long splitSize,
+ long inputSplitSize,
Review Comment:
I have forgot to rename these in my previous PR. During the review we changed from `splitSize` to `inputSplitSize` to highlight that this is for the input. We did the rename in the parent class (`RewriteGroupBase`), but I didn't do it here. Since this is a new class introduced by my previous PR, I felt that we should keep the naming consistent.
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2012966051
##########
core/src/main/java/org/apache/iceberg/actions/RewritePositionDeletesGroup.java:
##########
@@ -50,16 +53,16 @@ public RewritePositionDeletesGroup(
FileGroupInfo info,
List<PositionDeletesScanTask> tasks,
long writeMaxFileSize,
- long splitSize,
+ long inputSplitSize,
Review Comment:
Same as above, not sure we should do a rename here
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2016136447
##########
core/src/test/java/org/apache/iceberg/actions/TestBinPackRewriteFileGroupPlanner.java:
##########
@@ -0,0 +1,468 @@
+/*
+ * 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.actions;
+
+import static org.apache.iceberg.actions.BinPackRewriteFileGroupPlanner.MAX_FILE_SIZE_DEFAULT_RATIO;
+import static org.apache.iceberg.actions.RewriteDataFiles.REWRITE_JOB_ORDER;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.File;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.stream.Collectors;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.MockFileScanTask;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.TestBase;
+import org.apache.iceberg.TestTables;
+import org.apache.iceberg.actions.RewriteDataFiles.FileGroupInfo;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class TestBinPackRewriteFileGroupPlanner {
Review Comment:
Yes, most of them coming from the Spark code, but added missing tests too
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2013692297
##########
core/src/main/java/org/apache/iceberg/actions/RewritePositionDeletesGroup.java:
##########
@@ -50,16 +53,16 @@ public RewritePositionDeletesGroup(
FileGroupInfo info,
List<PositionDeletesScanTask> tasks,
long writeMaxFileSize,
- long splitSize,
+ long inputSplitSize,
Review Comment:
I have forgot to rename these in my previous PR. During the review we changed from splitSize to inputSplitSize to highlight that this is for the input. We did the rename in the parent class (RewriteGroupBase), but I didn't do it here. Since this is a new class introduced by my previous PR, I felt that we should keep the naming consistent.
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2016137775
##########
core/src/test/java/org/apache/iceberg/actions/TestBinPackRewriteFileGroupPlanner.java:
##########
@@ -0,0 +1,468 @@
+/*
+ * 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.actions;
+
+import static org.apache.iceberg.actions.BinPackRewriteFileGroupPlanner.MAX_FILE_SIZE_DEFAULT_RATIO;
+import static org.apache.iceberg.actions.RewriteDataFiles.REWRITE_JOB_ORDER;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.File;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.stream.Collectors;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.MockFileScanTask;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.TestBase;
+import org.apache.iceberg.TestTables;
+import org.apache.iceberg.actions.RewriteDataFiles.FileGroupInfo;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class TestBinPackRewriteFileGroupPlanner {
+ private static final Map<String, String> REWRITE_ALL =
+ ImmutableMap.of(BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true");
+
+ private static final DataFile FILE_1 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_2 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_3 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_4 = newDataFile("data_bucket=1", 11);
+ private static final DataFile FILE_5 = newDataFile("data_bucket=1", 11);
+ private static final DataFile FILE_6 = newDataFile("data_bucket=2", 50);
+
+ private static final Map<RewriteJobOrder, List<StructLike>> EXPECTED =
+ ImmutableMap.of(
+ RewriteJobOrder.FILES_DESC,
+ ImmutableList.of(FILE_1.partition(), FILE_4.partition(), FILE_6.partition()),
+ RewriteJobOrder.FILES_ASC,
+ ImmutableList.of(FILE_6.partition(), FILE_4.partition(), FILE_1.partition()),
+ RewriteJobOrder.BYTES_DESC,
+ ImmutableList.of(FILE_6.partition(), FILE_1.partition(), FILE_4.partition()),
+ RewriteJobOrder.BYTES_ASC,
+ ImmutableList.of(FILE_4.partition(), FILE_1.partition(), FILE_6.partition()));
+
+ @TempDir private File tableDir = null;
+ private TestTables.TestTable table = null;
+
+ @BeforeEach
+ public void setupTable() throws Exception {
+ this.table = TestTables.create(tableDir, "test", TestBase.SCHEMA, TestBase.SPEC, 3);
+ }
+
+ @AfterEach
+ public void cleanupTables() {
+ TestTables.clearTables();
+ }
+
+ @Test
+ void testPartitionedTable() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(ImmutableMap.of(BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups).hasSize(3);
+ assertThat(groups.stream().mapToInt(RewriteGroupBase::inputFileNum).sum()).isEqualTo(6);
+ }
+
+ @Test
+ void testUnpartitionedTable() {
+ table.updateSpec().removeField("data_bucket").commit();
+ table.refresh();
+ table
+ .newAppend()
+ .appendFile(newDataFile("", 10))
+ .appendFile(newDataFile("", 20))
+ .appendFile(newDataFile("", 30))
+ .commit();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.MIN_INPUT_FILES,
+ "1",
+ BinPackRewriteFileGroupPlanner.MIN_FILE_SIZE_BYTES,
+ "30"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.totalGroupCount()).isEqualTo(1);
+ assertThat(plan.groups().iterator().next().inputFileNum()).isEqualTo(2);
+ }
+
+ @Test
+ void testEmptyTable() {
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(REWRITE_ALL);
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(table.currentSnapshot()).as("Table must be empty").isNull();
+ assertThat(plan.totalGroupCount()).isZero();
+ }
+
+ @ParameterizedTest
+ @EnumSource(
+ value = RewriteJobOrder.class,
+ names = {"FILES_DESC", "FILES_ASC", "BYTES_DESC", "BYTES_ASC"})
+ void testJobOrder(RewriteJobOrder order) {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true", REWRITE_JOB_ORDER, order.name()));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups.stream().map(group -> group.info().partition()).collect(Collectors.toList()))
+ .isEqualTo(EXPECTED.get(order));
+ assertThat(plan.totalGroupCount()).isEqualTo(3);
+ EXPECTED.get(order).forEach(s -> assertThat(plan.groupsInPartition(s)).isEqualTo(1));
+ }
+
+ @Test
+ void testMaxGroupSize() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.MAX_FILE_GROUP_SIZE_BYTES,
+ "10"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.totalGroupCount()).isEqualTo(6);
+ assertThat(plan.groupsInPartition(FILE_1.partition())).isEqualTo(3);
+ assertThat(plan.groupsInPartition(FILE_4.partition())).isEqualTo(2);
+ assertThat(plan.groupsInPartition(FILE_6.partition())).isEqualTo(1);
+ }
+
+ @Test
+ void testFilter() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner =
+ new BinPackRewriteFileGroupPlanner(
+ table,
+ Expressions.or(
+ Expressions.equal(Expressions.bucket("data", 16), 0),
+ Expressions.equal(Expressions.bucket("data", 16), 2)));
+ planner.init(REWRITE_ALL);
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(plan.totalGroupCount()).isEqualTo(2);
Review Comment:
Added
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#issuecomment-2763228406
Merged to main.
Thanks for the review @stevenzwu and @RussellSpitzer!
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2015154046
##########
core/src/test/java/org/apache/iceberg/actions/TestBinPackRewriteFileGroupPlanner.java:
##########
@@ -0,0 +1,468 @@
+/*
+ * 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.actions;
+
+import static org.apache.iceberg.actions.BinPackRewriteFileGroupPlanner.MAX_FILE_SIZE_DEFAULT_RATIO;
+import static org.apache.iceberg.actions.RewriteDataFiles.REWRITE_JOB_ORDER;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.File;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.stream.Collectors;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.MockFileScanTask;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.TestBase;
+import org.apache.iceberg.TestTables;
+import org.apache.iceberg.actions.RewriteDataFiles.FileGroupInfo;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class TestBinPackRewriteFileGroupPlanner {
+ private static final Map<String, String> REWRITE_ALL =
+ ImmutableMap.of(BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true");
+
+ private static final DataFile FILE_1 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_2 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_3 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_4 = newDataFile("data_bucket=1", 11);
+ private static final DataFile FILE_5 = newDataFile("data_bucket=1", 11);
+ private static final DataFile FILE_6 = newDataFile("data_bucket=2", 50);
+
+ private static final Map<RewriteJobOrder, List<StructLike>> EXPECTED =
+ ImmutableMap.of(
+ RewriteJobOrder.FILES_DESC,
+ ImmutableList.of(FILE_1.partition(), FILE_4.partition(), FILE_6.partition()),
+ RewriteJobOrder.FILES_ASC,
+ ImmutableList.of(FILE_6.partition(), FILE_4.partition(), FILE_1.partition()),
+ RewriteJobOrder.BYTES_DESC,
+ ImmutableList.of(FILE_6.partition(), FILE_1.partition(), FILE_4.partition()),
+ RewriteJobOrder.BYTES_ASC,
+ ImmutableList.of(FILE_4.partition(), FILE_1.partition(), FILE_6.partition()));
+
+ @TempDir private File tableDir = null;
+ private TestTables.TestTable table = null;
+
+ @BeforeEach
+ public void setupTable() throws Exception {
+ this.table = TestTables.create(tableDir, "test", TestBase.SCHEMA, TestBase.SPEC, 3);
+ }
+
+ @AfterEach
+ public void cleanupTables() {
+ TestTables.clearTables();
+ }
+
+ @Test
+ void testPartitionedTable() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(ImmutableMap.of(BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups).hasSize(3);
+ assertThat(groups.stream().mapToInt(RewriteGroupBase::inputFileNum).sum()).isEqualTo(6);
+ }
+
+ @Test
+ void testUnpartitionedTable() {
+ table.updateSpec().removeField("data_bucket").commit();
+ table.refresh();
+ table
+ .newAppend()
+ .appendFile(newDataFile("", 10))
+ .appendFile(newDataFile("", 20))
+ .appendFile(newDataFile("", 30))
+ .commit();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.MIN_INPUT_FILES,
+ "1",
+ BinPackRewriteFileGroupPlanner.MIN_FILE_SIZE_BYTES,
+ "30"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.totalGroupCount()).isEqualTo(1);
+ assertThat(plan.groups().iterator().next().inputFileNum()).isEqualTo(2);
+ }
+
+ @Test
+ void testEmptyTable() {
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(REWRITE_ALL);
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(table.currentSnapshot()).as("Table must be empty").isNull();
+ assertThat(plan.totalGroupCount()).isZero();
+ }
+
+ @ParameterizedTest
+ @EnumSource(
+ value = RewriteJobOrder.class,
+ names = {"FILES_DESC", "FILES_ASC", "BYTES_DESC", "BYTES_ASC"})
+ void testJobOrder(RewriteJobOrder order) {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true", REWRITE_JOB_ORDER, order.name()));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups.stream().map(group -> group.info().partition()).collect(Collectors.toList()))
+ .isEqualTo(EXPECTED.get(order));
+ assertThat(plan.totalGroupCount()).isEqualTo(3);
+ EXPECTED.get(order).forEach(s -> assertThat(plan.groupsInPartition(s)).isEqualTo(1));
+ }
+
+ @Test
+ void testMaxGroupSize() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.MAX_FILE_GROUP_SIZE_BYTES,
+ "10"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.totalGroupCount()).isEqualTo(6);
+ assertThat(plan.groupsInPartition(FILE_1.partition())).isEqualTo(3);
+ assertThat(plan.groupsInPartition(FILE_4.partition())).isEqualTo(2);
+ assertThat(plan.groupsInPartition(FILE_6.partition())).isEqualTo(1);
+ }
+
+ @Test
+ void testFilter() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner =
+ new BinPackRewriteFileGroupPlanner(
+ table,
+ Expressions.or(
+ Expressions.equal(Expressions.bucket("data", 16), 0),
+ Expressions.equal(Expressions.bucket("data", 16), 2)));
+ planner.init(REWRITE_ALL);
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(plan.totalGroupCount()).isEqualTo(2);
+ assertThat(groups).hasSize(2);
+ assertThat(groups.stream().mapToLong(RewriteFileGroup::inputFileNum).sum()).isEqualTo(4);
+ }
+
+ @Test
+ void testMaxOutputFileSize() {
+ addFiles();
+ int targetFileSize = 10;
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.TARGET_FILE_SIZE_BYTES,
+ String.valueOf(targetFileSize)));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.groups().iterator().next().maxOutputFileSize())
+ .isGreaterThan(targetFileSize)
+ .isLessThan((long) (targetFileSize * MAX_FILE_SIZE_DEFAULT_RATIO));
+ }
+
+ @Test
+ void testExpectedOutputFiles() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.TARGET_FILE_SIZE_BYTES,
+ "21"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups).hasSize(3);
+ for (RewriteFileGroup group : groups) {
+ if (FILE_1.partition().equals(group.info().partition())) {
+ assertThat(group.expectedOutputFiles()).isEqualTo(2);
+ } else if (FILE_4.partition().equals(group.info().partition())) {
+ assertThat(group.expectedOutputFiles()).isEqualTo(1);
+ } else if (FILE_6.partition().equals(group.info().partition())) {
+ assertThat(group.expectedOutputFiles()).isEqualTo(3);
+ } else {
+ throw new IllegalStateException("Unexpected partition: " + group.info().partition());
+ }
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testOutputSpec(boolean specific) {
Review Comment:
nit: specific --> useOldSpecId
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2013698861
##########
core/src/main/java/org/apache/iceberg/actions/RewriteFileGroupPlanner.java:
##########
@@ -0,0 +1,310 @@
+/*
+ * 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.actions;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.actions.RewriteDataFiles.FileGroupInfo;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.ContentFileUtil;
+import org.apache.iceberg.util.PropertyUtil;
+import org.apache.iceberg.util.StructLikeMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Groups specified data files in the {@link Table} into {@link RewriteFileGroup}s. The files are
+ * grouped by partitions based on their size using fix sized bins. Extends {@link
+ * SizeBasedFileRewritePlanner} with delete file number and delete ratio thresholds and job {@link
+ * RewriteDataFiles#REWRITE_JOB_ORDER} handling.
+ */
+public class RewriteFileGroupPlanner
Review Comment:
Agreed. Renamed to `BinPackRewriteFileGroupPlanner`
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#issuecomment-2753796490
Thanks @RussellSpitzer for the review!
I have tried to answer and address your questions!
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2012945454
##########
core/src/main/java/org/apache/iceberg/actions/RewriteFileGroupPlanner.java:
##########
@@ -0,0 +1,310 @@
+/*
+ * 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.actions;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.actions.RewriteDataFiles.FileGroupInfo;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.ContentFileUtil;
+import org.apache.iceberg.util.PropertyUtil;
+import org.apache.iceberg.util.StructLikeMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Groups specified data files in the {@link Table} into {@link RewriteFileGroup}s. The files are
+ * grouped by partitions based on their size using fix sized bins. Extends {@link
+ * SizeBasedFileRewritePlanner} with delete file number and delete ratio thresholds and job {@link
+ * RewriteDataFiles#REWRITE_JOB_ORDER} handling.
+ */
+public class RewriteFileGroupPlanner
+ extends SizeBasedFileRewritePlanner<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> {
+ /**
+ * The minimum number of deletes that needs to be associated with a data file for it to be
+ * considered for rewriting. If a data file has this number of deletes or more, it will be
+ * rewritten regardless of its file size determined by {@link #MIN_FILE_SIZE_BYTES} and {@link
+ * #MAX_FILE_SIZE_BYTES}. If a file group contains a file that satisfies this condition, the file
+ * group will be rewritten regardless of the number of files in the file group determined by
+ * {@link #MIN_INPUT_FILES}.
+ *
+ * <p>Defaults to Integer.MAX_VALUE, which means this feature is not enabled by default.
+ */
+ public static final String DELETE_FILE_THRESHOLD = "delete-file-threshold";
+
+ public static final int DELETE_FILE_THRESHOLD_DEFAULT = Integer.MAX_VALUE;
+
+ /**
+ * The minimum deletion ratio that needs to be associated with a data file for it to be considered
+ * for rewriting. If the deletion ratio of a data file is greater than or equal to this value, it
+ * will be rewritten regardless of its file size determined by {@link #MIN_FILE_SIZE_BYTES} and
+ * {@link #MAX_FILE_SIZE_BYTES}. If a file group contains a file that satisfies this condition,
+ * the file group will be rewritten regardless of the number of files in the file group determined
+ * by {@link #MIN_INPUT_FILES}.
+ *
+ * <p>Defaults to 0.3, which means that if the deletion ratio of a file reaches or exceeds 30%, it
+ * may trigger the rewriting operation.
+ */
+ public static final String DELETE_RATIO_THRESHOLD = "delete-ratio-threshold";
+
+ public static final double DELETE_RATIO_THRESHOLD_DEFAULT = 0.3;
+
+ private static final Logger LOG = LoggerFactory.getLogger(RewriteFileGroupPlanner.class);
+
+ private final Expression filter;
+ private final Long snapshotId;
+ private final boolean caseSensitive;
+
+ private int deleteFileThreshold;
+ private double deleteRatioThreshold;
+ private RewriteJobOrder rewriteJobOrder;
+
+ public RewriteFileGroupPlanner(Table table) {
+ this(table, Expressions.alwaysTrue());
+ }
+
+ public RewriteFileGroupPlanner(Table table, Expression filter) {
+ this(
+ table,
+ filter,
+ table.currentSnapshot() != null ? table.currentSnapshot().snapshotId() : null,
+ false);
+ }
+
+ /**
+ * Creates the planner for the given table.
+ *
+ * @param table to plan for
+ * @param filter used to remove files from the plan
+ * @param snapshotId used as a basis for planning - should be used as starting snapshot id at
+ * commit time when replacing the files
+ * @param caseSensitive property used for scanning
+ */
+ public RewriteFileGroupPlanner(
+ Table table, Expression filter, Long snapshotId, boolean caseSensitive) {
+ super(table);
+ this.filter = filter;
+ this.snapshotId = snapshotId;
+ this.caseSensitive = caseSensitive;
+ }
+
+ @Override
+ public Set<String> validOptions() {
+ return ImmutableSet.<String>builder()
+ .addAll(super.validOptions())
+ .add(DELETE_FILE_THRESHOLD)
+ .add(DELETE_RATIO_THRESHOLD)
+ .add(RewriteDataFiles.REWRITE_JOB_ORDER)
+ .build();
+ }
+
+ @Override
+ public void init(Map<String, String> options) {
+ super.init(options);
+ this.deleteFileThreshold = deleteFileThreshold(options);
+ this.deleteRatioThreshold = deleteRatioThreshold(options);
+ this.rewriteJobOrder =
+ RewriteJobOrder.fromName(
+ PropertyUtil.propertyAsString(
+ options,
+ RewriteDataFiles.REWRITE_JOB_ORDER,
Review Comment:
Do these properties belong here now?
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2012934021
##########
core/src/main/java/org/apache/iceberg/actions/RewriteFileGroupPlanner.java:
##########
@@ -0,0 +1,310 @@
+/*
+ * 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.actions;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.actions.RewriteDataFiles.FileGroupInfo;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.ContentFileUtil;
+import org.apache.iceberg.util.PropertyUtil;
+import org.apache.iceberg.util.StructLikeMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Groups specified data files in the {@link Table} into {@link RewriteFileGroup}s. The files are
+ * grouped by partitions based on their size using fix sized bins. Extends {@link
+ * SizeBasedFileRewritePlanner} with delete file number and delete ratio thresholds and job {@link
+ * RewriteDataFiles#REWRITE_JOB_ORDER} handling.
+ */
+public class RewriteFileGroupPlanner
+ extends SizeBasedFileRewritePlanner<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> {
+ /**
+ * The minimum number of deletes that needs to be associated with a data file for it to be
+ * considered for rewriting. If a data file has this number of deletes or more, it will be
+ * rewritten regardless of its file size determined by {@link #MIN_FILE_SIZE_BYTES} and {@link
+ * #MAX_FILE_SIZE_BYTES}. If a file group contains a file that satisfies this condition, the file
+ * group will be rewritten regardless of the number of files in the file group determined by
+ * {@link #MIN_INPUT_FILES}.
+ *
+ * <p>Defaults to Integer.MAX_VALUE, which means this feature is not enabled by default.
+ */
+ public static final String DELETE_FILE_THRESHOLD = "delete-file-threshold";
+
+ public static final int DELETE_FILE_THRESHOLD_DEFAULT = Integer.MAX_VALUE;
+
+ /**
+ * The minimum deletion ratio that needs to be associated with a data file for it to be considered
Review Comment:
nit: I assume this is a copy but the wording doesn't make sense
Should be like
"The minimum ratio of deleted rows to live rows in a data file for the file to be considered for rewriting."
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2012929572
##########
core/src/main/java/org/apache/iceberg/actions/RewriteFileGroup.java:
##########
@@ -48,14 +51,14 @@ public RewriteFileGroup(
List<FileScanTask> fileScanTasks,
int outputSpecId,
long writeMaxFileSize,
- long splitSize,
+ long inputSplitSize,
Review Comment:
Why rename here?
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2015095511
##########
core/src/main/java/org/apache/iceberg/actions/BinPackRewriteFileGroupPlanner.java:
##########
@@ -0,0 +1,310 @@
+/*
+ * 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.actions;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.actions.RewriteDataFiles.FileGroupInfo;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.ContentFileUtil;
+import org.apache.iceberg.util.PropertyUtil;
+import org.apache.iceberg.util.StructLikeMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Groups specified data files in the {@link Table} into {@link RewriteFileGroup}s. The files are
+ * grouped by partitions based on their size using fix sized bins. Extends {@link
+ * SizeBasedFileRewritePlanner} with delete file number and delete ratio thresholds and job {@link
+ * RewriteDataFiles#REWRITE_JOB_ORDER} handling.
+ */
+public class BinPackRewriteFileGroupPlanner
+ extends SizeBasedFileRewritePlanner<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> {
+ /**
+ * The minimum number of deletes that needs to be associated with a data file for it to be
+ * considered for rewriting. If a data file has this number of deletes or more, it will be
+ * rewritten regardless of its file size determined by {@link #MIN_FILE_SIZE_BYTES} and {@link
+ * #MAX_FILE_SIZE_BYTES}. If a file group contains a file that satisfies this condition, the file
+ * group will be rewritten regardless of the number of files in the file group determined by
+ * {@link #MIN_INPUT_FILES}.
+ *
+ * <p>Defaults to Integer.MAX_VALUE, which means this feature is not enabled by default.
+ */
+ public static final String DELETE_FILE_THRESHOLD = "delete-file-threshold";
+
+ public static final int DELETE_FILE_THRESHOLD_DEFAULT = Integer.MAX_VALUE;
+
+ /**
+ * The minimum deletion ratio that needs to be associated with a data file for it to be considered
+ * for rewriting. If the deletion ratio of a data file is greater than or equal to this value, it
+ * will be rewritten regardless of its file size determined by {@link #MIN_FILE_SIZE_BYTES} and
+ * {@link #MAX_FILE_SIZE_BYTES}. If a file group contains a file that satisfies this condition,
+ * the file group will be rewritten regardless of the number of files in the file group determined
+ * by {@link #MIN_INPUT_FILES}.
+ *
+ * <p>Defaults to 0.3, which means that if the deletion ratio of a file reaches or exceeds 30%, it
+ * may trigger the rewriting operation.
+ */
+ public static final String DELETE_RATIO_THRESHOLD = "delete-ratio-threshold";
+
+ public static final double DELETE_RATIO_THRESHOLD_DEFAULT = 0.3;
+
+ private static final Logger LOG = LoggerFactory.getLogger(BinPackRewriteFileGroupPlanner.class);
+
+ private final Expression filter;
+ private final Long snapshotId;
+ private final boolean caseSensitive;
+
+ private int deleteFileThreshold;
+ private double deleteRatioThreshold;
+ private RewriteJobOrder rewriteJobOrder;
+
+ public BinPackRewriteFileGroupPlanner(Table table) {
+ this(table, Expressions.alwaysTrue());
+ }
+
+ public BinPackRewriteFileGroupPlanner(Table table, Expression filter) {
+ this(
+ table,
+ filter,
+ table.currentSnapshot() != null ? table.currentSnapshot().snapshotId() : null,
+ false);
+ }
+
+ /**
+ * Creates the planner for the given table.
+ *
+ * @param table to plan for
+ * @param filter used to remove files from the plan
+ * @param snapshotId used as a basis for planning - should be used as starting snapshot id at
Review Comment:
nit on Javadoc
```
@param snapshotId a snapshot ID used for planning and as the starting snapshot id for commit validation when replacing the files
```
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2014975086
##########
core/src/main/java/org/apache/iceberg/actions/SizeBasedFileRewritePlanner.java:
##########
@@ -0,0 +1,361 @@
+/*
+ * 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.actions;
+
+import java.math.RoundingMode;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.ContentScanTask;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.math.LongMath;
+import org.apache.iceberg.util.BinPacking;
+import org.apache.iceberg.util.PropertyUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A file rewrite planner that determines which files to rewrite based on their size.
+ *
+ * <p>If files are smaller than the {@link #MIN_FILE_SIZE_BYTES} threshold or larger than the {@link
+ * #MAX_FILE_SIZE_BYTES} threshold, they are considered targets for being rewritten.
+ *
+ * <p>Once selected, files are grouped based on the {@link BinPacking bin-packing algorithm} into
+ * groups of no more than {@link #MAX_FILE_GROUP_SIZE_BYTES}. Groups will be actually rewritten if
+ * they contain more than {@link #MIN_INPUT_FILES} or if they would produce at least one file of
+ * {@link #TARGET_FILE_SIZE_BYTES}.
+ *
+ * <p>Note that implementations may add extra conditions for selecting files or filtering groups.
+ */
+public abstract class SizeBasedFileRewritePlanner<
+ I,
+ T extends ContentScanTask<F>,
+ F extends ContentFile<F>,
+ G extends RewriteGroupBase<I, T, F>>
+ implements FileRewritePlanner<I, T, F, G> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SizeBasedFileRewritePlanner.class);
+
+ /** The target output file size that this file rewriter will attempt to generate. */
+ public static final String TARGET_FILE_SIZE_BYTES = "target-file-size-bytes";
+
+ /**
+ * Controls which files will be considered for rewriting. Files with sizes under this threshold
+ * will be considered for rewriting regardless of any other criteria.
+ *
+ * <p>Defaults to 75% of the target file size.
+ */
+ public static final String MIN_FILE_SIZE_BYTES = "min-file-size-bytes";
+
+ public static final double MIN_FILE_SIZE_DEFAULT_RATIO = 0.75;
+
+ /**
+ * Controls which files will be considered for rewriting. Files with sizes above this threshold
+ * will be considered for rewriting regardless of any other criteria.
+ *
+ * <p>Defaults to 180% of the target file size.
+ */
+ public static final String MAX_FILE_SIZE_BYTES = "max-file-size-bytes";
+
+ public static final double MAX_FILE_SIZE_DEFAULT_RATIO = 1.80;
+
+ /**
+ * Any file group exceeding this number of files will be rewritten regardless of other criteria.
+ * This config ensures file groups that contain many files are compacted even if the total size of
+ * that group is less than the target file size. This can also be thought of as the maximum number
+ * of wrongly sized files that could remain in a partition after rewriting.
+ */
+ public static final String MIN_INPUT_FILES = "min-input-files";
+
+ public static final int MIN_INPUT_FILES_DEFAULT = 5;
+
+ /** Overrides other options and forces rewriting of all provided files. */
+ public static final String REWRITE_ALL = "rewrite-all";
+
+ public static final boolean REWRITE_ALL_DEFAULT = false;
+
+ /**
+ * This option controls the largest amount of data that should be rewritten in a single file
+ * group. It helps with breaking down the rewriting of very large partitions which may not be
+ * rewritable otherwise due to the resource constraints of the cluster. For example, a sort-based
+ * rewrite may not scale to TB-sized partitions, and those partitions need to be worked on in
+ * small subsections to avoid exhaustion of resources.
+ */
+ public static final String MAX_FILE_GROUP_SIZE_BYTES = "max-file-group-size-bytes";
+
+ public static final long MAX_FILE_GROUP_SIZE_BYTES_DEFAULT = 100L * 1024 * 1024 * 1024; // 100 GB
+
+ private static final long SPLIT_OVERHEAD = 5L * 1024;
+
+ private final Table table;
+ private long targetFileSize;
+ private long minFileSize;
+ private long maxFileSize;
+ private int minInputFiles;
+ private boolean rewriteAll;
+ private long maxGroupSize;
+ private int outputSpecId;
+
+ protected SizeBasedFileRewritePlanner(Table table) {
+ this.table = table;
+ }
+
+ /** Expected target file size before configuration. */
+ protected abstract long defaultTargetFileSize();
+
+ /** Additional filter for tasks before grouping. */
+ protected abstract Iterable<T> filterFiles(Iterable<T> tasks);
+
+ /** Additional filter for groups. */
+ protected abstract Iterable<List<T>> filterFileGroups(List<List<T>> groups);
+
+ @Override
+ public Set<String> validOptions() {
+ return ImmutableSet.of(
+ TARGET_FILE_SIZE_BYTES,
+ MIN_FILE_SIZE_BYTES,
+ MAX_FILE_SIZE_BYTES,
+ MIN_INPUT_FILES,
+ REWRITE_ALL,
+ MAX_FILE_GROUP_SIZE_BYTES);
+ }
+
+ @Override
+ public void init(Map<String, String> options) {
+ Map<String, Long> sizeThresholds = sizeThresholds(options);
+ this.targetFileSize = sizeThresholds.get(TARGET_FILE_SIZE_BYTES);
+ this.minFileSize = sizeThresholds.get(MIN_FILE_SIZE_BYTES);
+ this.maxFileSize = sizeThresholds.get(MAX_FILE_SIZE_BYTES);
+ this.minInputFiles = minInputFiles(options);
+ this.rewriteAll = rewriteAll(options);
+ this.maxGroupSize = maxGroupSize(options);
+ this.outputSpecId = outputSpecId(options);
+
+ if (rewriteAll) {
+ LOG.info("Configured to rewrite all provided files in table {}", table.name());
+ }
+ }
+
+ protected Table table() {
+ return table;
+ }
+
+ protected boolean wronglySized(T task) {
Review Comment:
I know the method name is copied from existing code. I don't know if we are entertaining maybe some renaming to make it easier to read.
e.g., this one can be called `outsideDesiredFileSizeRange`.
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2014782759
##########
core/src/main/java/org/apache/iceberg/actions/RewritePositionDeletesGroupPlanner.java:
##########
@@ -0,0 +1,221 @@
+/*
+ * 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.actions;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.MetadataTableType;
+import org.apache.iceberg.MetadataTableUtils;
+import org.apache.iceberg.Partitioning;
+import org.apache.iceberg.PositionDeletesScanTask;
+import org.apache.iceberg.PositionDeletesTable;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.actions.RewritePositionDeleteFiles.FileGroupInfo;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.PartitionUtil;
+import org.apache.iceberg.util.PropertyUtil;
+import org.apache.iceberg.util.StructLikeMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Groups specified delete files in the {@link Table} into {@link RewritePositionDeletesGroup}s.
+ * These will be grouped by partitions based on their size using fix sized bins. Extends the {@link
+ * SizeBasedFileRewritePlanner} with {@link RewritePositionDeleteFiles#REWRITE_JOB_ORDER} handling.
+ */
+public class RewritePositionDeletesGroupPlanner
Review Comment:
probably `RewritePositionDeletesPlanner` is more accurate
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2012937314
##########
core/src/main/java/org/apache/iceberg/actions/RewriteFileGroupPlanner.java:
##########
@@ -0,0 +1,310 @@
+/*
+ * 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.actions;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.actions.RewriteDataFiles.FileGroupInfo;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.ContentFileUtil;
+import org.apache.iceberg.util.PropertyUtil;
+import org.apache.iceberg.util.StructLikeMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Groups specified data files in the {@link Table} into {@link RewriteFileGroup}s. The files are
+ * grouped by partitions based on their size using fix sized bins. Extends {@link
+ * SizeBasedFileRewritePlanner} with delete file number and delete ratio thresholds and job {@link
+ * RewriteDataFiles#REWRITE_JOB_ORDER} handling.
+ */
+public class RewriteFileGroupPlanner
+ extends SizeBasedFileRewritePlanner<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> {
+ /**
+ * The minimum number of deletes that needs to be associated with a data file for it to be
+ * considered for rewriting. If a data file has this number of deletes or more, it will be
+ * rewritten regardless of its file size determined by {@link #MIN_FILE_SIZE_BYTES} and {@link
+ * #MAX_FILE_SIZE_BYTES}. If a file group contains a file that satisfies this condition, the file
+ * group will be rewritten regardless of the number of files in the file group determined by
+ * {@link #MIN_INPUT_FILES}.
+ *
+ * <p>Defaults to Integer.MAX_VALUE, which means this feature is not enabled by default.
+ */
+ public static final String DELETE_FILE_THRESHOLD = "delete-file-threshold";
+
+ public static final int DELETE_FILE_THRESHOLD_DEFAULT = Integer.MAX_VALUE;
+
+ /**
+ * The minimum deletion ratio that needs to be associated with a data file for it to be considered
Review Comment:
really this whole paragraph should remove the term "Deletion ratio" or at least define it :)
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2017052085
##########
core/src/main/java/org/apache/iceberg/actions/RewritePositionDeletesGroupPlanner.java:
##########
@@ -0,0 +1,221 @@
+/*
+ * 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.actions;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.MetadataTableType;
+import org.apache.iceberg.MetadataTableUtils;
+import org.apache.iceberg.Partitioning;
+import org.apache.iceberg.PositionDeletesScanTask;
+import org.apache.iceberg.PositionDeletesTable;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.actions.RewritePositionDeleteFiles.FileGroupInfo;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.PartitionUtil;
+import org.apache.iceberg.util.PropertyUtil;
+import org.apache.iceberg.util.StructLikeMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Groups specified delete files in the {@link Table} into {@link RewritePositionDeletesGroup}s.
+ * These will be grouped by partitions based on their size using fix sized bins. Extends the {@link
+ * SizeBasedFileRewritePlanner} with {@link RewritePositionDeleteFiles#REWRITE_JOB_ORDER} handling.
+ */
+public class RewritePositionDeletesGroupPlanner
+ extends SizeBasedFileRewritePlanner<
+ FileGroupInfo, PositionDeletesScanTask, DeleteFile, RewritePositionDeletesGroup> {
+ private static final Logger LOG =
+ LoggerFactory.getLogger(RewritePositionDeletesGroupPlanner.class);
+
+ private final Expression filter;
+ private final boolean caseSensitive;
+ private RewriteJobOrder rewriteJobOrder;
+
+ public RewritePositionDeletesGroupPlanner(Table table) {
+ this(table, Expressions.alwaysTrue(), false);
+ }
+
+ /**
+ * Creates the planner for the given table.
+ *
+ * @param table to plan for
+ * @param filter used to remove files from the plan
+ * @param caseSensitive property used for scanning
+ */
+ public RewritePositionDeletesGroupPlanner(Table table, Expression filter, boolean caseSensitive) {
+ super(table);
+ this.caseSensitive = caseSensitive;
+ this.filter = filter;
+ }
+
+ @Override
+ public Set<String> validOptions() {
+ return ImmutableSet.<String>builder()
+ .addAll(super.validOptions())
+ .add(RewritePositionDeleteFiles.REWRITE_JOB_ORDER)
+ .build();
+ }
+
+ @Override
+ public void init(Map<String, String> options) {
+ super.init(options);
+ this.rewriteJobOrder =
+ RewriteJobOrder.fromName(
+ PropertyUtil.propertyAsString(
+ options,
+ RewritePositionDeleteFiles.REWRITE_JOB_ORDER,
+ RewritePositionDeleteFiles.REWRITE_JOB_ORDER_DEFAULT));
+ }
+
+ @Override
+ public FileRewritePlan<
+ FileGroupInfo, PositionDeletesScanTask, DeleteFile, RewritePositionDeletesGroup>
+ plan() {
+ StructLikeMap<List<List<PositionDeletesScanTask>>> plan = planFileGroups();
+ RewriteExecutionContext ctx = new RewriteExecutionContext();
+ Stream<RewritePositionDeletesGroup> groups =
+ plan.entrySet().stream()
+ .filter(e -> !e.getValue().isEmpty())
+ .flatMap(
+ e -> {
+ StructLike partition = e.getKey();
+ List<List<PositionDeletesScanTask>> scanGroups = e.getValue();
+ return scanGroups.stream()
+ .map(
+ tasks -> {
+ long inputSize = inputSize(tasks);
+ return newRewriteGroup(
+ ctx,
+ partition,
+ tasks,
+ inputSplitSize(inputSize),
+ expectedOutputFiles(inputSize));
+ });
+ })
+ .sorted(RewritePositionDeletesGroup.comparator(rewriteJobOrder));
+ Map<StructLike, Integer> groupsInPartition = plan.transformValues(List::size);
+ int totalGroupCount = groupsInPartition.values().stream().reduce(Integer::sum).orElse(0);
+ return new FileRewritePlan<>(
+ CloseableIterable.of(groups.collect(Collectors.toList())),
+ totalGroupCount,
+ groupsInPartition);
+ }
+
+ @Override
+ protected Iterable<PositionDeletesScanTask> filterFiles(Iterable<PositionDeletesScanTask> tasks) {
+ return Iterables.filter(tasks, this::wronglySized);
+ }
+
+ @Override
+ protected Iterable<List<PositionDeletesScanTask>> filterFileGroups(
+ List<List<PositionDeletesScanTask>> groups) {
+ return Iterables.filter(
+ groups, group -> enoughInputFiles(group) || enoughContent(group) || tooMuchContent(group));
+ }
+
+ @Override
+ protected long defaultTargetFileSize() {
+ return PropertyUtil.propertyAsLong(
+ table().properties(),
+ TableProperties.DELETE_TARGET_FILE_SIZE_BYTES,
+ TableProperties.DELETE_TARGET_FILE_SIZE_BYTES_DEFAULT);
+ }
+
+ private StructLikeMap<List<List<PositionDeletesScanTask>>> planFileGroups() {
+ Table deletesTable =
+ MetadataTableUtils.createMetadataTableInstance(table(), MetadataTableType.POSITION_DELETES);
+ CloseableIterable<PositionDeletesScanTask> fileTasks = planFiles(deletesTable);
+
+ try {
+ Types.StructType partitionType = Partitioning.partitionType(deletesTable);
+ StructLikeMap<List<PositionDeletesScanTask>> fileTasksByPartition =
+ groupByPartition(partitionType, fileTasks);
+ return fileTasksByPartition.transformValues(
+ tasks -> ImmutableList.copyOf(planFileGroups(tasks)));
+ } finally {
+ try {
+ fileTasks.close();
+ } catch (IOException io) {
+ LOG.error("Cannot properly close file iterable while planning for rewrite", io);
+ }
+ }
+ }
+
+ private CloseableIterable<PositionDeletesScanTask> planFiles(Table deletesTable) {
+ PositionDeletesTable.PositionDeletesBatchScan scan =
+ (PositionDeletesTable.PositionDeletesBatchScan) deletesTable.newBatchScan();
+ return CloseableIterable.transform(
+ scan.baseTableFilter(filter).caseSensitive(caseSensitive).ignoreResiduals().planFiles(),
+ PositionDeletesScanTask.class::cast);
+ }
+
+ private StructLikeMap<List<PositionDeletesScanTask>> groupByPartition(
Review Comment:
sounds good.
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2015155240
##########
core/src/test/java/org/apache/iceberg/actions/TestBinPackRewriteFileGroupPlanner.java:
##########
@@ -0,0 +1,468 @@
+/*
+ * 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.actions;
+
+import static org.apache.iceberg.actions.BinPackRewriteFileGroupPlanner.MAX_FILE_SIZE_DEFAULT_RATIO;
+import static org.apache.iceberg.actions.RewriteDataFiles.REWRITE_JOB_ORDER;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.File;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.stream.Collectors;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.MockFileScanTask;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.TestBase;
+import org.apache.iceberg.TestTables;
+import org.apache.iceberg.actions.RewriteDataFiles.FileGroupInfo;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class TestBinPackRewriteFileGroupPlanner {
+ private static final Map<String, String> REWRITE_ALL =
+ ImmutableMap.of(BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true");
+
+ private static final DataFile FILE_1 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_2 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_3 = newDataFile("data_bucket=0", 10);
+ private static final DataFile FILE_4 = newDataFile("data_bucket=1", 11);
+ private static final DataFile FILE_5 = newDataFile("data_bucket=1", 11);
+ private static final DataFile FILE_6 = newDataFile("data_bucket=2", 50);
+
+ private static final Map<RewriteJobOrder, List<StructLike>> EXPECTED =
+ ImmutableMap.of(
+ RewriteJobOrder.FILES_DESC,
+ ImmutableList.of(FILE_1.partition(), FILE_4.partition(), FILE_6.partition()),
+ RewriteJobOrder.FILES_ASC,
+ ImmutableList.of(FILE_6.partition(), FILE_4.partition(), FILE_1.partition()),
+ RewriteJobOrder.BYTES_DESC,
+ ImmutableList.of(FILE_6.partition(), FILE_1.partition(), FILE_4.partition()),
+ RewriteJobOrder.BYTES_ASC,
+ ImmutableList.of(FILE_4.partition(), FILE_1.partition(), FILE_6.partition()));
+
+ @TempDir private File tableDir = null;
+ private TestTables.TestTable table = null;
+
+ @BeforeEach
+ public void setupTable() throws Exception {
+ this.table = TestTables.create(tableDir, "test", TestBase.SCHEMA, TestBase.SPEC, 3);
+ }
+
+ @AfterEach
+ public void cleanupTables() {
+ TestTables.clearTables();
+ }
+
+ @Test
+ void testPartitionedTable() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(ImmutableMap.of(BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups).hasSize(3);
+ assertThat(groups.stream().mapToInt(RewriteGroupBase::inputFileNum).sum()).isEqualTo(6);
+ }
+
+ @Test
+ void testUnpartitionedTable() {
+ table.updateSpec().removeField("data_bucket").commit();
+ table.refresh();
+ table
+ .newAppend()
+ .appendFile(newDataFile("", 10))
+ .appendFile(newDataFile("", 20))
+ .appendFile(newDataFile("", 30))
+ .commit();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.MIN_INPUT_FILES,
+ "1",
+ BinPackRewriteFileGroupPlanner.MIN_FILE_SIZE_BYTES,
+ "30"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.totalGroupCount()).isEqualTo(1);
+ assertThat(plan.groups().iterator().next().inputFileNum()).isEqualTo(2);
+ }
+
+ @Test
+ void testEmptyTable() {
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(REWRITE_ALL);
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(table.currentSnapshot()).as("Table must be empty").isNull();
+ assertThat(plan.totalGroupCount()).isZero();
+ }
+
+ @ParameterizedTest
+ @EnumSource(
+ value = RewriteJobOrder.class,
+ names = {"FILES_DESC", "FILES_ASC", "BYTES_DESC", "BYTES_ASC"})
+ void testJobOrder(RewriteJobOrder order) {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL, "true", REWRITE_JOB_ORDER, order.name()));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups.stream().map(group -> group.info().partition()).collect(Collectors.toList()))
+ .isEqualTo(EXPECTED.get(order));
+ assertThat(plan.totalGroupCount()).isEqualTo(3);
+ EXPECTED.get(order).forEach(s -> assertThat(plan.groupsInPartition(s)).isEqualTo(1));
+ }
+
+ @Test
+ void testMaxGroupSize() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.MAX_FILE_GROUP_SIZE_BYTES,
+ "10"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.totalGroupCount()).isEqualTo(6);
+ assertThat(plan.groupsInPartition(FILE_1.partition())).isEqualTo(3);
+ assertThat(plan.groupsInPartition(FILE_4.partition())).isEqualTo(2);
+ assertThat(plan.groupsInPartition(FILE_6.partition())).isEqualTo(1);
+ }
+
+ @Test
+ void testFilter() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner =
+ new BinPackRewriteFileGroupPlanner(
+ table,
+ Expressions.or(
+ Expressions.equal(Expressions.bucket("data", 16), 0),
+ Expressions.equal(Expressions.bucket("data", 16), 2)));
+ planner.init(REWRITE_ALL);
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(plan.totalGroupCount()).isEqualTo(2);
+ assertThat(groups).hasSize(2);
+ assertThat(groups.stream().mapToLong(RewriteFileGroup::inputFileNum).sum()).isEqualTo(4);
+ }
+
+ @Test
+ void testMaxOutputFileSize() {
+ addFiles();
+ int targetFileSize = 10;
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.TARGET_FILE_SIZE_BYTES,
+ String.valueOf(targetFileSize)));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.groups().iterator().next().maxOutputFileSize())
+ .isGreaterThan(targetFileSize)
+ .isLessThan((long) (targetFileSize * MAX_FILE_SIZE_DEFAULT_RATIO));
+ }
+
+ @Test
+ void testExpectedOutputFiles() {
+ addFiles();
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(
+ ImmutableMap.of(
+ BinPackRewriteFileGroupPlanner.REWRITE_ALL,
+ "true",
+ BinPackRewriteFileGroupPlanner.TARGET_FILE_SIZE_BYTES,
+ "21"));
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ List<RewriteFileGroup> groups = Lists.newArrayList(plan.groups().iterator());
+ assertThat(groups).hasSize(3);
+ for (RewriteFileGroup group : groups) {
+ if (FILE_1.partition().equals(group.info().partition())) {
+ assertThat(group.expectedOutputFiles()).isEqualTo(2);
+ } else if (FILE_4.partition().equals(group.info().partition())) {
+ assertThat(group.expectedOutputFiles()).isEqualTo(1);
+ } else if (FILE_6.partition().equals(group.info().partition())) {
+ assertThat(group.expectedOutputFiles()).isEqualTo(3);
+ } else {
+ throw new IllegalStateException("Unexpected partition: " + group.info().partition());
+ }
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testOutputSpec(boolean specific) {
+ addFiles();
+ int oldSpecId = table.spec().specId();
+ table.updateSpec().removeField("data_bucket").commit();
+ table.newAppend().appendFile(newDataFile("", 10)).commit();
+ table.refresh();
+ int newSpecId = table.spec().specId();
+
+ Map<String, String> options = Maps.newHashMap(REWRITE_ALL);
+ if (specific) {
+ options.put(RewriteDataFiles.OUTPUT_SPEC_ID, String.valueOf(oldSpecId));
+ }
+
+ BinPackRewriteFileGroupPlanner planner = new BinPackRewriteFileGroupPlanner(table);
+ planner.init(options);
+
+ FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan = planner.plan();
+
+ assertThat(plan.groups().iterator().next().outputSpecId())
Review Comment:
nit: validate every group?
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2015139989
##########
core/src/test/java/org/apache/iceberg/actions/TestBinPackRewriteFileGroupPlanner.java:
##########
@@ -0,0 +1,468 @@
+/*
+ * 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.actions;
+
+import static org.apache.iceberg.actions.BinPackRewriteFileGroupPlanner.MAX_FILE_SIZE_DEFAULT_RATIO;
+import static org.apache.iceberg.actions.RewriteDataFiles.REWRITE_JOB_ORDER;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.File;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.stream.Collectors;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.MockFileScanTask;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.TestBase;
+import org.apache.iceberg.TestTables;
+import org.apache.iceberg.actions.RewriteDataFiles.FileGroupInfo;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class TestBinPackRewriteFileGroupPlanner {
Review Comment:
I assume most tests are refactored from the spark bin pack action test?
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2013719350
##########
core/src/main/java/org/apache/iceberg/actions/RewriteFileGroupPlanner.java:
##########
@@ -0,0 +1,310 @@
+/*
+ * 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.actions;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.actions.RewriteDataFiles.FileGroupInfo;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.ContentFileUtil;
+import org.apache.iceberg.util.PropertyUtil;
+import org.apache.iceberg.util.StructLikeMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Groups specified data files in the {@link Table} into {@link RewriteFileGroup}s. The files are
+ * grouped by partitions based on their size using fix sized bins. Extends {@link
+ * SizeBasedFileRewritePlanner} with delete file number and delete ratio thresholds and job {@link
+ * RewriteDataFiles#REWRITE_JOB_ORDER} handling.
+ */
+public class RewriteFileGroupPlanner
+ extends SizeBasedFileRewritePlanner<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> {
+ /**
+ * The minimum number of deletes that needs to be associated with a data file for it to be
+ * considered for rewriting. If a data file has this number of deletes or more, it will be
+ * rewritten regardless of its file size determined by {@link #MIN_FILE_SIZE_BYTES} and {@link
+ * #MAX_FILE_SIZE_BYTES}. If a file group contains a file that satisfies this condition, the file
+ * group will be rewritten regardless of the number of files in the file group determined by
+ * {@link #MIN_INPUT_FILES}.
+ *
+ * <p>Defaults to Integer.MAX_VALUE, which means this feature is not enabled by default.
+ */
+ public static final String DELETE_FILE_THRESHOLD = "delete-file-threshold";
+
+ public static final int DELETE_FILE_THRESHOLD_DEFAULT = Integer.MAX_VALUE;
+
+ /**
+ * The minimum deletion ratio that needs to be associated with a data file for it to be considered
Review Comment:
Yeah. Copied from `SizeBasedDataRewriter` which added this by #12148 and #11825
Rewritten a bit:
```
/**
* The percentage of the deleted records in a data file that enforces a data file rewrite. If the
* deletion ratio of a data file is greater or equal than this value, it will be rewritten
* regardless of its file size determined by {@link #MIN_FILE_SIZE_BYTES} and {@link
* #MAX_FILE_SIZE_BYTES}. If a file group contains a file that satisfies this condition, the file
* group will be rewritten regardless of the number of files in the file group determined by
* {@link #MIN_INPUT_FILES}.
*
* <p>Defaults to 0.3, which means that if the number of deleted records in a file reaches or
* exceeds 30%, it may trigger the rewriting operation.
*/
```
If you have more suggestions, I would be happy to apply 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] Core: FileRewritePlanner implementation [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2015078649
##########
core/src/main/java/org/apache/iceberg/actions/BinPackRewriteFileGroupPlanner.java:
##########
@@ -0,0 +1,310 @@
+/*
+ * 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.actions;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.RewriteJobOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.actions.RewriteDataFiles.FileGroupInfo;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.ContentFileUtil;
+import org.apache.iceberg.util.PropertyUtil;
+import org.apache.iceberg.util.StructLikeMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Groups specified data files in the {@link Table} into {@link RewriteFileGroup}s. The files are
+ * grouped by partitions based on their size using fix sized bins. Extends {@link
+ * SizeBasedFileRewritePlanner} with delete file number and delete ratio thresholds and job {@link
+ * RewriteDataFiles#REWRITE_JOB_ORDER} handling.
+ */
+public class BinPackRewriteFileGroupPlanner
+ extends SizeBasedFileRewritePlanner<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> {
+ /**
+ * The minimum number of deletes that needs to be associated with a data file for it to be
+ * considered for rewriting. If a data file has this number of deletes or more, it will be
+ * rewritten regardless of its file size determined by {@link #MIN_FILE_SIZE_BYTES} and {@link
+ * #MAX_FILE_SIZE_BYTES}. If a file group contains a file that satisfies this condition, the file
+ * group will be rewritten regardless of the number of files in the file group determined by
+ * {@link #MIN_INPUT_FILES}.
+ *
+ * <p>Defaults to Integer.MAX_VALUE, which means this feature is not enabled by default.
+ */
+ public static final String DELETE_FILE_THRESHOLD = "delete-file-threshold";
Review Comment:
I know this is refactored from existing code. this config name doesn't seem the most intuitive. `delete-file-threshold` could create an impression of delete file size threshold (in bytes). I think it meant `deleted-rows-threshold`. But it is probably too late to change it for compatibility reason.
--
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] Core: FileRewritePlanner implementation [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12493:
URL: https://github.com/apache/iceberg/pull/12493#discussion_r2016083801
##########
core/src/main/java/org/apache/iceberg/actions/SizeBasedDataRewriter.java:
##########
@@ -49,15 +56,15 @@ public abstract class SizeBasedDataRewriter extends SizeBasedFileRewriter<FileSc
public static final int DELETE_FILE_THRESHOLD_DEFAULT = Integer.MAX_VALUE;
/**
- * The minimum deletion ratio that needs to be associated with a data file for it to be considered
- * for rewriting. If the deletion ratio of a data file is greater than or equal to this value, it
- * will be rewritten regardless of its file size determined by {@link #MIN_FILE_SIZE_BYTES} and
- * {@link #MAX_FILE_SIZE_BYTES}. If a file group contains a file that satisfies this condition,
- * the file group will be rewritten regardless of the number of files in the file group determined
- * by {@link #MIN_INPUT_FILES}.
+ * The percentage of the deleted records in a data file that enforces a data file rewrite. If the
+ * deletion ratio of a data file is greater or equal than this value, it will be rewritten
+ * regardless of its file size determined by {@link #MIN_FILE_SIZE_BYTES} and {@link
+ * #MAX_FILE_SIZE_BYTES}. If a file group contains a file that satisfies this condition, the file
+ * group will be rewritten regardless of the number of files in the file group determined by
+ * {@link #MIN_INPUT_FILES}.
*
- * <p>Defaults to 0.3, which means that if the deletion ratio of a file reaches or exceeds 30%, it
- * may trigger the rewriting operation.
+ * <p>Defaults to 0.3, which means that if the number of deleted records in a file reaches or
+ * exceeds 30%, it may trigger the rewriting operation.
Review Comment:
Done here, and in the `BinPackRewriteFilePlanner` too
--
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