You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "Zouxxyy (via GitHub)" <gi...@apache.org> on 2023/12/06 02:03:50 UTC
[PR] [spark] Optimize compact procedure in non-sort scenario [incubator-paimon]
Zouxxyy opened a new pull request, #2457:
URL: https://github.com/apache/incubator-paimon/pull/2457
<!-- Please specify the module before the PR name: [core] ... or [flink] ... -->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: close #xxx
<!-- What is the purpose of the change -->
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Optimize compact procedure in non-sort scenario [incubator-paimon]
Posted by "leaves12138 (via GitHub)" <gi...@apache.org>.
leaves12138 commented on code in PR #2457:
URL: https://github.com/apache/incubator-paimon/pull/2457#discussion_r1418405072
##########
paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java:
##########
@@ -124,28 +148,155 @@ private boolean execute(
FileStoreTable table,
String sortType,
List<String> sortColumns,
- @Nullable String filter) {
- CoreOptions coreOptions = table.store().options();
-
- // sort only works with bucket=-1 yet
- if (!TableSorter.OrderType.of(sortType).equals(TableSorter.OrderType.NONE)) {
- if (!(table instanceof AppendOnlyFileStoreTable) || coreOptions.bucket() != -1) {
- throw new UnsupportedOperationException(
- "Spark compact with sort_type "
- + sortType
- + " only support unaware-bucket append-only table yet.");
+ @Nullable String partitions) {
+ table = table.copy(Collections.singletonMap(CoreOptions.WRITE_ONLY.key(), "false"));
+ BucketMode bucketMode = table.bucketMode();
+ TableSorter.OrderType orderType = TableSorter.OrderType.of(sortType);
+
+ if (orderType.equals(TableSorter.OrderType.NONE)) {
+ JavaSparkContext javaSparkContext = new JavaSparkContext(spark().sparkContext());
+ Predicate filter =
+ StringUtils.isBlank(partitions)
+ ? null
+ : ParameterUtils.getPartitionFilter(
+ ParameterUtils.getPartitions(partitions), table.rowType());
+ switch (bucketMode) {
+ case FIXED:
+ case DYNAMIC:
+ compactAwareBucketTable(table, filter, javaSparkContext);
+ break;
+ case UNAWARE:
+ compactUnAwareBucketTable(table, filter, javaSparkContext);
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ "Spark compact with " + bucketMode + " is not support yet.");
+ }
+ } else {
+ switch (bucketMode) {
+ case UNAWARE:
+ sortCompactUnAwareBucketTable(table, orderType, sortColumns, partitions);
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ "Spark compact with sort_type "
+ + sortType
+ + " only support unaware-bucket append-only table yet.");
+ }
+ }
+ return true;
+ }
+
+ private void compactAwareBucketTable(
+ FileStoreTable table, @Nullable Predicate filter, JavaSparkContext javaSparkContext) {
+ InnerTableScan scan = table.newScan();
+ if (filter != null) {
+ scan.withFilter(filter);
+ }
+
+ List<Split> splits = scan.plan().splits();
+ if (splits.isEmpty()) {
+ return;
+ }
+
+ BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+ JavaRDD<CommitMessage> commitMessageJavaRDD =
+ javaSparkContext
+ .parallelize(splits)
+ .mapPartitions(
+ (FlatMapFunction<Iterator<Split>, CommitMessage>)
+ splitIterator -> {
+ IOManager ioManager = SparkUtils.createIOManager();
+ BatchTableWrite write = writeBuilder.newWrite();
+ write.withIOManager(ioManager);
Review Comment:
If one partition bucket have two splits, this may cause an error.
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Optimize compact procedure in non-sort scenario [incubator-paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #2457:
URL: https://github.com/apache/incubator-paimon/pull/2457
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Optimize compact procedure in non-sort scenario [incubator-paimon]
Posted by "leaves12138 (via GitHub)" <gi...@apache.org>.
leaves12138 commented on code in PR #2457:
URL: https://github.com/apache/incubator-paimon/pull/2457#discussion_r1418247793
##########
paimon-common/src/main/java/org/apache/paimon/utils/ParameterUtils.java:
##########
@@ -34,6 +38,14 @@ public static List<Map<String, String>> getPartitions(String... partitionStrings
return partitions;
}
+ public static Predicate getPartitionFilter(
Review Comment:
ParameterUtils is used for handling input string to other format, like Map List<Map> List etc. This class is not suitable for construct Predicate with input(Map, RowType).
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org