You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by lz...@apache.org on 2022/07/08 03:08:26 UTC
[flink-table-store] branch master updated: [FLINK-28442] SplitGenerator should use Ordered Packing
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-table-store.git
The following commit(s) were added to refs/heads/master by this push:
new 38257d65 [FLINK-28442] SplitGenerator should use Ordered Packing
38257d65 is described below
commit 38257d6595596c8e7f9c8060ba4ea0b252594139
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Fri Jul 8 11:08:22 2022 +0800
[FLINK-28442] SplitGenerator should use Ordered Packing
This closes #206
---
.../apache/flink/table/store/utils/BinPacking.java | 96 ----------------------
.../flink/table/store/utils/OrderedPacking.java | 42 ++++++----
.../table/source/AppendOnlySplitGenerator.java | 4 +-
.../table/source/MergeTreeSplitGenerator.java | 12 +--
.../store/table/source/SplitGeneratorTest.java | 26 +++---
5 files changed, 46 insertions(+), 134 deletions(-)
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/BinPacking.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/BinPacking.java
deleted file mode 100644
index 0bde5f76..00000000
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/BinPacking.java
+++ /dev/null
@@ -1,96 +0,0 @@
-/*
- * 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.flink.table.store.utils;
-
-import org.apache.flink.annotation.Internal;
-
-import java.util.ArrayList;
-import java.util.Deque;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.function.Function;
-
-/**
- * A bin packing implementation.
- *
- * <p>NOTE: Copied from Flink.
- */
-@Internal
-public class BinPacking {
- private BinPacking() {}
-
- public static <T> List<List<T>> pack(
- Iterable<T> items, Function<T, Long> weightFunc, long targetWeight) {
- List<List<T>> packed = new ArrayList<>();
- Deque<Bin<T>> bins = new LinkedList<>();
-
- for (T item : items) {
- long weight = weightFunc.apply(item);
- Bin<T> bin = findBin(bins, weight);
-
- if (bin != null) {
- bin.add(item, weight);
- } else {
- bin = new Bin<>(targetWeight);
- bin.add(item, weight);
- bins.addLast(bin);
-
- // avoid n * n algorithm complexity
- if (bins.size() > 10) {
- packed.add(bins.removeFirst().items());
- }
- }
- }
-
- bins.forEach(bin -> packed.add(bin.items));
- return packed;
- }
-
- private static <T> Bin<T> findBin(Iterable<Bin<T>> bins, long weight) {
- for (Bin<T> bin : bins) {
- if (bin.canAdd(weight)) {
- return bin;
- }
- }
- return null;
- }
-
- private static class Bin<T> {
- private final long targetWeight;
- private final List<T> items = new ArrayList<>();
- private long binWeight = 0L;
-
- Bin(long targetWeight) {
- this.targetWeight = targetWeight;
- }
-
- List<T> items() {
- return items;
- }
-
- boolean canAdd(long weight) {
- return binWeight + weight <= targetWeight;
- }
-
- void add(T item, long weight) {
- this.binWeight += weight;
- items.add(item);
- }
- }
-}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/AppendOnlySplitGenerator.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/OrderedPacking.java
similarity index 50%
copy from flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/AppendOnlySplitGenerator.java
copy to flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/OrderedPacking.java
index bdb6912d..bd3295c6 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/AppendOnlySplitGenerator.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/OrderedPacking.java
@@ -16,28 +16,38 @@
* limitations under the License.
*/
-package org.apache.flink.table.store.table.source;
-
-import org.apache.flink.table.store.file.data.DataFileMeta;
-import org.apache.flink.table.store.utils.BinPacking;
+package org.apache.flink.table.store.utils;
+import java.util.ArrayList;
import java.util.List;
import java.util.function.Function;
-/** Append only implementation of {@link SplitGenerator}. */
-public class AppendOnlySplitGenerator implements SplitGenerator {
+/** Ordered packing for input items. */
+public class OrderedPacking {
+ private OrderedPacking() {}
- private final long targetSplitSize;
- private final long openFileCost;
+ public static <T> List<List<T>> pack(
+ Iterable<T> items, Function<T, Long> weightFunc, long targetWeight) {
+ List<List<T>> packed = new ArrayList<>();
- public AppendOnlySplitGenerator(long targetSplitSize, long openFileCost) {
- this.targetSplitSize = targetSplitSize;
- this.openFileCost = openFileCost;
- }
+ List<T> binItems = new ArrayList<>();
+ long binWeight = 0L;
+
+ for (T item : items) {
+ long weight = weightFunc.apply(item);
+ if (binWeight + weight > targetWeight) {
+ packed.add(binItems);
+ binItems = new ArrayList<>();
+ binWeight = 0;
+ }
+
+ binWeight += weight;
+ binItems.add(item);
+ }
- @Override
- public List<List<DataFileMeta>> split(List<DataFileMeta> files) {
- Function<DataFileMeta, Long> weightFunc = file -> Math.max(file.fileSize(), openFileCost);
- return BinPacking.pack(files, weightFunc, targetSplitSize);
+ if (binItems.size() > 0) {
+ packed.add(binItems);
+ }
+ return packed;
}
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/AppendOnlySplitGenerator.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/AppendOnlySplitGenerator.java
index bdb6912d..e6312378 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/AppendOnlySplitGenerator.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/AppendOnlySplitGenerator.java
@@ -19,7 +19,7 @@
package org.apache.flink.table.store.table.source;
import org.apache.flink.table.store.file.data.DataFileMeta;
-import org.apache.flink.table.store.utils.BinPacking;
+import org.apache.flink.table.store.utils.OrderedPacking;
import java.util.List;
import java.util.function.Function;
@@ -38,6 +38,6 @@ public class AppendOnlySplitGenerator implements SplitGenerator {
@Override
public List<List<DataFileMeta>> split(List<DataFileMeta> files) {
Function<DataFileMeta, Long> weightFunc = file -> Math.max(file.fileSize(), openFileCost);
- return BinPacking.pack(files, weightFunc, targetSplitSize);
+ return OrderedPacking.pack(files, weightFunc, targetSplitSize);
}
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/MergeTreeSplitGenerator.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/MergeTreeSplitGenerator.java
index 48d2a471..5a79a778 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/MergeTreeSplitGenerator.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/MergeTreeSplitGenerator.java
@@ -22,7 +22,7 @@ import org.apache.flink.table.data.RowData;
import org.apache.flink.table.store.file.data.DataFileMeta;
import org.apache.flink.table.store.file.mergetree.SortedRun;
import org.apache.flink.table.store.file.mergetree.compact.IntervalPartition;
-import org.apache.flink.table.store.utils.BinPacking;
+import org.apache.flink.table.store.utils.OrderedPacking;
import java.util.ArrayList;
import java.util.Comparator;
@@ -53,7 +53,7 @@ public class MergeTreeSplitGenerator implements SplitGenerator {
* into multiple splits. The generation has one constraint: files with intersected key
* ranges (within one section) must go to the same split. Therefore, the files are first to go
* through the interval partition algorithm to generate sections and then through the
- * BinPack algorithm. Note that the item to be packed here is each section, the bin capacity
+ * OrderedPack algorithm. Note that the item to be packed here is each section, the capacity
* is denoted as the targetSplitSize, and the final number of the bins is the number of
* splits generated.
*
@@ -64,7 +64,7 @@ public class MergeTreeSplitGenerator implements SplitGenerator {
* - section3: [5, 180], [5, 190]
* - section4: [200, 600], [210, 700]
*
- * After BinPack, section1 and section2 will be put into one bin (split), so the final result will be:
+ * After OrderedPack, section1 and section2 will be put into one bin (split), so the final result will be:
* - split1: [1, 2] [3, 4]
* - split2: [5, 180] [5,190]
* - split3: [200, 600] [210, 700]
@@ -73,13 +73,13 @@ public class MergeTreeSplitGenerator implements SplitGenerator {
new IntervalPartition(files, keyComparator)
.partition().stream().map(this::flatRun).collect(Collectors.toList());
- return binPackSplits(sections);
+ return packSplits(sections);
}
- private List<List<DataFileMeta>> binPackSplits(List<List<DataFileMeta>> sections) {
+ private List<List<DataFileMeta>> packSplits(List<List<DataFileMeta>> sections) {
Function<List<DataFileMeta>, Long> weightFunc =
file -> Math.max(totalSize(file), openFileCost);
- return BinPacking.pack(sections, weightFunc, targetSplitSize).stream()
+ return OrderedPacking.pack(sections, weightFunc, targetSplitSize).stream()
.map(this::flatFiles)
.collect(Collectors.toList());
}
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/source/SplitGeneratorTest.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/source/SplitGeneratorTest.java
index 9a8feb0b..e8fb66f9 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/source/SplitGeneratorTest.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/source/SplitGeneratorTest.java
@@ -48,21 +48,22 @@ public class SplitGeneratorTest {
public void testAppend() {
assertThat(toNames(new AppendOnlySplitGenerator(40, 2).split(files)))
.containsExactlyInAnyOrder(
- Arrays.asList("1", "2", "5"),
+ Arrays.asList("1", "2"),
Collections.singletonList("3"),
- Collections.singletonList("4"),
+ Arrays.asList("4", "5"),
Collections.singletonList("6"));
- assertThat(toNames(new AppendOnlySplitGenerator(50, 2).split(files)))
+ assertThat(toNames(new AppendOnlySplitGenerator(70, 2).split(files)))
.containsExactlyInAnyOrder(
- Arrays.asList("1", "2", "4"),
- Arrays.asList("3", "5"),
+ Arrays.asList("1", "2", "3"),
+ Arrays.asList("4", "5"),
Collections.singletonList("6"));
- assertThat(toNames(new AppendOnlySplitGenerator(50, 10).split(files)))
+ assertThat(toNames(new AppendOnlySplitGenerator(40, 20).split(files)))
.containsExactlyInAnyOrder(
- Arrays.asList("1", "2", "4"),
+ Arrays.asList("1", "2"),
Collections.singletonList("3"),
+ Collections.singletonList("4"),
Collections.singletonList("5"),
Collections.singletonList("6"));
}
@@ -70,16 +71,13 @@ public class SplitGeneratorTest {
@Test
public void testMergeTree() {
Comparator<RowData> comparator = Comparator.comparingInt(o -> o.getInt(0));
- assertThat(toNames(new MergeTreeSplitGenerator(comparator, 40, 2).split(files)))
+ assertThat(toNames(new MergeTreeSplitGenerator(comparator, 100, 2).split(files)))
.containsExactlyInAnyOrder(
- Arrays.asList("1", "2", "5"),
- Arrays.asList("4", "3"),
- Collections.singletonList("6"));
+ Arrays.asList("1", "2", "4", "3", "5"), Collections.singletonList("6"));
- assertThat(toNames(new MergeTreeSplitGenerator(comparator, 40, 20).split(files)))
+ assertThat(toNames(new MergeTreeSplitGenerator(comparator, 100, 30).split(files)))
.containsExactlyInAnyOrder(
- Arrays.asList("1", "2"),
- Arrays.asList("4", "3"),
+ Arrays.asList("1", "2", "4", "3"),
Collections.singletonList("5"),
Collections.singletonList("6"));
}