You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by GitBox <gi...@apache.org> on 2021/08/02 18:18:14 UTC
[GitHub] [iceberg] karuppayya opened a new pull request #2921: Add an argument to parallelize Expire snapshot procedure
karuppayya opened a new pull request #2921:
URL: https://github.com/apache/iceberg/pull/2921
--
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
[GitHub] [iceberg] RussellSpitzer commented on a change in pull request #2921: Add an argument to parallelize Expire snapshot procedure
Posted by GitBox <gi...@apache.org>.
RussellSpitzer commented on a change in pull request #2921:
URL: https://github.com/apache/iceberg/pull/2921#discussion_r718794871
##########
File path: spark3/src/main/java/org/apache/iceberg/spark/procedures/ExpireSnapshotsProcedure.java
##########
@@ -79,6 +85,7 @@ public StructType outputType() {
Identifier tableIdent = toIdentifier(args.getString(0), PARAMETERS[0].name());
Long olderThanMillis = args.isNullAt(1) ? null : DateTimeUtil.microsToMillis(args.getLong(1));
Integer retainLastNum = args.isNullAt(2) ? null : args.getInt(2);
+ Integer maxConcurrentDeletes = args.isNullAt(3) ? null : args.getInt(3);
Review comment:
I think the only thing left is we should check that maxCurrentDeletes is null or > 0, after that I think we are good to merge
--
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
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2921: Add an argument to parallelize Expire snapshot procedure
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2921:
URL: https://github.com/apache/iceberg/pull/2921#discussion_r707682020
##########
File path: spark3/src/main/java/org/apache/iceberg/spark/procedures/ExpireSnapshotsProcedure.java
##########
@@ -90,7 +96,14 @@ public StructType outputType() {
if (retainLastNum != null) {
action.retainLast(retainLastNum);
}
-
+ if (numThreads != null && numThreads > 0) {
+ action.executeDeleteWith(Executors.newFixedThreadPool(numThreads, runnable -> {
+ Thread thread = new Thread(runnable);
+ thread.setName("expire-snapshot-" + deleteThreadsIndex.getAndIncrement());
Review comment:
nit: `snapshots`?
--
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
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2921: Add an argument to parallelize Expire snapshot procedure
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2921:
URL: https://github.com/apache/iceberg/pull/2921#discussion_r707681680
##########
File path: spark3/src/main/java/org/apache/iceberg/spark/procedures/ExpireSnapshotsProcedure.java
##########
@@ -42,7 +44,8 @@
private static final ProcedureParameter[] PARAMETERS = new ProcedureParameter[] {
ProcedureParameter.required("table", DataTypes.StringType),
ProcedureParameter.optional("older_than", DataTypes.TimestampType),
- ProcedureParameter.optional("retain_last", DataTypes.IntegerType)
+ ProcedureParameter.optional("retain_last", DataTypes.IntegerType),
+ ProcedureParameter.optional("num_parallelism", DataTypes.IntegerType)
Review comment:
I agree with @kbendick that the name isn't very descriptive and parallelism can be mistaken with the number of shuffle partitions in Spark.
We could call it `delete_thread_pool_size` or `num_delete_threads`.
Thoughts, @karuppayya @kbendick @RussellSpitzer @raptond @flyrain @jackye1995?
--
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
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2921: Add an argument to parallelize Expire snapshot procedure
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2921:
URL: https://github.com/apache/iceberg/pull/2921#discussion_r707684171
##########
File path: spark3/src/main/java/org/apache/iceberg/spark/procedures/ExpireSnapshotsProcedure.java
##########
@@ -51,6 +54,8 @@
new StructField("deleted_manifest_lists_count", DataTypes.LongType, true, Metadata.empty())
});
+ private AtomicInteger deleteThreadsIndex = new AtomicInteger(0);
Review comment:
Nit: could be final or even converted into a local variable?
--
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
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2921: Add an argument to parallelize Expire snapshot procedure
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2921:
URL: https://github.com/apache/iceberg/pull/2921#discussion_r707682247
##########
File path: spark3/src/main/java/org/apache/iceberg/spark/procedures/ExpireSnapshotsProcedure.java
##########
@@ -90,7 +96,14 @@ public StructType outputType() {
if (retainLastNum != null) {
action.retainLast(retainLastNum);
}
-
+ if (numThreads != null && numThreads > 0) {
+ action.executeDeleteWith(Executors.newFixedThreadPool(numThreads, runnable -> {
Review comment:
Do we need to wrap this into `MoreExecutors.getExitingExecutorService`?
--
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
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2921: Add an argument to parallelize Expire snapshot procedure
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2921:
URL: https://github.com/apache/iceberg/pull/2921#discussion_r707682994
##########
File path: spark3/src/main/java/org/apache/iceberg/spark/procedures/ExpireSnapshotsProcedure.java
##########
@@ -90,7 +96,14 @@ public StructType outputType() {
if (retainLastNum != null) {
action.retainLast(retainLastNum);
}
-
+ if (numThreads != null && numThreads > 0) {
+ action.executeDeleteWith(Executors.newFixedThreadPool(numThreads, runnable -> {
+ Thread thread = new Thread(runnable);
+ thread.setName("expire-snapshot-" + deleteThreadsIndex.getAndIncrement());
+ thread.setDaemon(true); // daemon threads will be terminated abruptly when the JVM exits
+ return thread;
+ }));
+ }
ExpireSnapshots.Result result = action.execute();
Review comment:
nit: I'd probably keep the extra empty line after this if block 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
[GitHub] [iceberg] RussellSpitzer commented on a change in pull request #2921: Add an argument to parallelize Expire snapshot procedure
Posted by GitBox <gi...@apache.org>.
RussellSpitzer commented on a change in pull request #2921:
URL: https://github.com/apache/iceberg/pull/2921#discussion_r707686694
##########
File path: spark3/src/main/java/org/apache/iceberg/spark/procedures/ExpireSnapshotsProcedure.java
##########
@@ -90,7 +96,14 @@ public StructType outputType() {
if (retainLastNum != null) {
action.retainLast(retainLastNum);
}
-
+ if (numThreads != null && numThreads > 0) {
Review comment:
I'd just default this to 1 instead of null and do a precondition on numThreads > 0
Currently I believe this is just ignoring invalid numThreads right?
--
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
[GitHub] [iceberg] jackye1995 commented on a change in pull request #2921: Add an argument to parallelize Expire snapshot procedure
Posted by GitBox <gi...@apache.org>.
jackye1995 commented on a change in pull request #2921:
URL: https://github.com/apache/iceberg/pull/2921#discussion_r707786029
##########
File path: spark3/src/main/java/org/apache/iceberg/spark/procedures/ExpireSnapshotsProcedure.java
##########
@@ -90,7 +96,14 @@ public StructType outputType() {
if (retainLastNum != null) {
action.retainLast(retainLastNum);
}
-
+ if (numThreads != null && numThreads > 0) {
+ action.executeDeleteWith(Executors.newFixedThreadPool(numThreads, runnable -> {
+ Thread thread = new Thread(runnable);
+ thread.setName("expire-snapshot-" + deleteThreadsIndex.getAndIncrement());
+ thread.setDaemon(true); // daemon threads will be terminated abruptly when the JVM exits
Review comment:
if we use exiting executor service, I think this line is not needed? Based on guava doc https://guava.dev/releases/19.0/api/docs/com/google/common/util/concurrent/MoreExecutors.html#getExitingExecutorService(java.util.concurrent.ThreadPoolExecutor):
> Converts the given ThreadPoolExecutor into an ExecutorService that exits when the application is complete. It does so by using daemon threads and adding a shutdown hook to wait for their completion.
--
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
[GitHub] [iceberg] aokolnychyi commented on pull request #2921: Add an argument to parallelize Expire snapshot procedure
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on pull request #2921:
URL: https://github.com/apache/iceberg/pull/2921#issuecomment-918582832
I think this will be a valuable addition. We should probably fix the arg name and also add a test where this argument is set. I know it will be hard to test the number of threads used but at least we should ensure there is no exception.
--
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
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2921: Add an argument to parallelize Expire snapshot procedure
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2921:
URL: https://github.com/apache/iceberg/pull/2921#discussion_r707682784
##########
File path: spark3/src/main/java/org/apache/iceberg/spark/procedures/ExpireSnapshotsProcedure.java
##########
@@ -90,7 +96,14 @@ public StructType outputType() {
if (retainLastNum != null) {
action.retainLast(retainLastNum);
}
-
Review comment:
Could you keep extra empty lines around the new if block to match the styling of existing if blocks?
--
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
[GitHub] [iceberg] jackye1995 commented on a change in pull request #2921: Add an argument to parallelize Expire snapshot procedure
Posted by GitBox <gi...@apache.org>.
jackye1995 commented on a change in pull request #2921:
URL: https://github.com/apache/iceberg/pull/2921#discussion_r707785204
##########
File path: spark3/src/main/java/org/apache/iceberg/spark/procedures/ExpireSnapshotsProcedure.java
##########
@@ -90,7 +96,14 @@ public StructType outputType() {
if (retainLastNum != null) {
action.retainLast(retainLastNum);
}
-
+ if (numThreads != null && numThreads > 0) {
Review comment:
I think keeping the null value is the correct behavior, because if `executeDeleteWith` is not set, then the execution will happen in the current thread. Otherwise this will create a new thread pool of size 1 and use that for execution.
--
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
[GitHub] [iceberg] kbendick commented on a change in pull request #2921: Add an argument to parallelize Expire snapshot procedure
Posted by GitBox <gi...@apache.org>.
kbendick commented on a change in pull request #2921:
URL: https://github.com/apache/iceberg/pull/2921#discussion_r683760308
##########
File path: spark3/src/main/java/org/apache/iceberg/spark/procedures/ExpireSnapshotsProcedure.java
##########
@@ -42,7 +44,8 @@
private static final ProcedureParameter[] PARAMETERS = new ProcedureParameter[] {
ProcedureParameter.required("table", DataTypes.StringType),
ProcedureParameter.optional("older_than", DataTypes.TimestampType),
- ProcedureParameter.optional("retain_last", DataTypes.IntegerType)
+ ProcedureParameter.optional("retain_last", DataTypes.IntegerType),
+ ProcedureParameter.optional("num_parallelism", DataTypes.IntegerType)
Review comment:
Nit (non-blocking): This parameter name is a bit confusing for me. Perhaps simply `num_threads` would do? It is indeed difficult to think of a name that does not evoke the normal spark task parallelism. Maybe even just `parallelism` would do?
--
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
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2921: Add an argument to parallelize Expire snapshot procedure
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2921:
URL: https://github.com/apache/iceberg/pull/2921#discussion_r707691201
##########
File path: spark3/src/main/java/org/apache/iceberg/spark/procedures/ExpireSnapshotsProcedure.java
##########
@@ -90,7 +96,14 @@ public StructType outputType() {
if (retainLastNum != null) {
action.retainLast(retainLastNum);
}
-
+ if (numThreads != null && numThreads > 0) {
Review comment:
+1 on adding validation.
I would consider keeping the null value, though, to allow the underlying action to default it to whatever it wants. I think we do that for other params.
--
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
[GitHub] [iceberg] RussellSpitzer merged pull request #2921: Add an argument to parallelize Expire snapshot procedure
Posted by GitBox <gi...@apache.org>.
RussellSpitzer merged pull request #2921:
URL: https://github.com/apache/iceberg/pull/2921
--
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
[GitHub] [iceberg] RussellSpitzer commented on a change in pull request #2921: Add an argument to parallelize Expire snapshot procedure
Posted by GitBox <gi...@apache.org>.
RussellSpitzer commented on a change in pull request #2921:
URL: https://github.com/apache/iceberg/pull/2921#discussion_r718794871
##########
File path: spark3/src/main/java/org/apache/iceberg/spark/procedures/ExpireSnapshotsProcedure.java
##########
@@ -79,6 +85,7 @@ public StructType outputType() {
Identifier tableIdent = toIdentifier(args.getString(0), PARAMETERS[0].name());
Long olderThanMillis = args.isNullAt(1) ? null : DateTimeUtil.microsToMillis(args.getLong(1));
Integer retainLastNum = args.isNullAt(2) ? null : args.getInt(2);
+ Integer maxConcurrentDeletes = args.isNullAt(3) ? null : args.getInt(3);
Review comment:
I think the only thing left is we should check that maxCurrentDeletes is null or > 0, after that I think we are good to merge
--
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
[GitHub] [iceberg] RussellSpitzer commented on a change in pull request #2921: Add an argument to parallelize Expire snapshot procedure
Posted by GitBox <gi...@apache.org>.
RussellSpitzer commented on a change in pull request #2921:
URL: https://github.com/apache/iceberg/pull/2921#discussion_r707682598
##########
File path: spark3/src/main/java/org/apache/iceberg/spark/procedures/ExpireSnapshotsProcedure.java
##########
@@ -42,7 +44,8 @@
private static final ProcedureParameter[] PARAMETERS = new ProcedureParameter[] {
ProcedureParameter.required("table", DataTypes.StringType),
ProcedureParameter.optional("older_than", DataTypes.TimestampType),
- ProcedureParameter.optional("retain_last", DataTypes.IntegerType)
+ ProcedureParameter.optional("retain_last", DataTypes.IntegerType),
+ ProcedureParameter.optional("num_parallelism", DataTypes.IntegerType)
Review comment:
concurrent_deletes?
--
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
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2921: Add an argument to parallelize Expire snapshot procedure
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2921:
URL: https://github.com/apache/iceberg/pull/2921#discussion_r707684723
##########
File path: spark3/src/main/java/org/apache/iceberg/spark/procedures/ExpireSnapshotsProcedure.java
##########
@@ -90,7 +96,14 @@ public StructType outputType() {
if (retainLastNum != null) {
action.retainLast(retainLastNum);
}
-
+ if (numThreads != null && numThreads > 0) {
+ action.executeDeleteWith(Executors.newFixedThreadPool(numThreads, runnable -> {
Review comment:
I'd probably move the initialization of a thread pool into a separate method that would accept a size.
Something similar to what @RussellSpitzer did in `BaseRewriteDataFilesSparkAction`.
--
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
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2921: Add an argument to parallelize Expire snapshot procedure
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2921:
URL: https://github.com/apache/iceberg/pull/2921#discussion_r707687552
##########
File path: spark3/src/main/java/org/apache/iceberg/spark/procedures/ExpireSnapshotsProcedure.java
##########
@@ -42,7 +44,8 @@
private static final ProcedureParameter[] PARAMETERS = new ProcedureParameter[] {
ProcedureParameter.required("table", DataTypes.StringType),
ProcedureParameter.optional("older_than", DataTypes.TimestampType),
- ProcedureParameter.optional("retain_last", DataTypes.IntegerType)
+ ProcedureParameter.optional("retain_last", DataTypes.IntegerType),
+ ProcedureParameter.optional("num_parallelism", DataTypes.IntegerType)
Review comment:
I like that, it is shorter. We could prepend `max_concurrent_deletes` to match naming in the rewrite action.
--
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
[GitHub] [iceberg] RussellSpitzer commented on a change in pull request #2921: Add an argument to parallelize Expire snapshot procedure
Posted by GitBox <gi...@apache.org>.
RussellSpitzer commented on a change in pull request #2921:
URL: https://github.com/apache/iceberg/pull/2921#discussion_r707686168
##########
File path: spark3/src/main/java/org/apache/iceberg/spark/procedures/ExpireSnapshotsProcedure.java
##########
@@ -90,7 +96,14 @@ public StructType outputType() {
if (retainLastNum != null) {
action.retainLast(retainLastNum);
}
-
+ if (numThreads != null && numThreads > 0) {
+ action.executeDeleteWith(Executors.newFixedThreadPool(numThreads, runnable -> {
Review comment:
``` private ExecutorService rewriteService() {
return MoreExecutors.getExitingExecutorService(
(ThreadPoolExecutor) Executors.newFixedThreadPool(
maxConcurrentFileGroupRewrites,
new ThreadFactoryBuilder()
.setNameFormat("Rewrite-Service-%d")
.build()));
}```
--
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