This is an automated email from the ASF dual-hosted git repository.
roryqi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 0237d79b73 [#13726] feat(policy): Support Iceberg rewrite strategy and
sort order (#13727)
0237d79b73 is described below
commit 0237d79b737a688659481505c32b7dee5f49411b
Author: MaSai <[email protected]>
AuthorDate: Sat Oct 10 16:16:27 2026 +0800
[#13726] feat(policy): Support Iceberg rewrite strategy and sort order
(#13727)
### What changes were proposed in this pull request?
- Add typed policy fields `rewriteStrategy` / `sortOrder` on
`system_iceberg_compaction`
- Emit them as rules `job.strategy` / `job.sort-order` (not under
`job.options.*`)
- Propagate through `CompactionJobContext` →
`GravitinoCompactionJobAdapter` job template args `strategy` /
`sort_order`
- Keep `rewriteOptions` as the Iceberg `options` map only
- Update docs and unit tests
### Why are the changes needed?
The optimizer previously hardcoded `strategy=binpack` and empty
`sort_order`. Sort/z-order compaction cannot be configured via
`rewriteOptions` because Iceberg treats those as top-level procedure
parameters, separate from the `options` map.
Fix: #13726
### Does this PR introduce _any_ user-facing change?
Yes:
- New optional content fields: `rewriteStrategy` (default `binpack`),
`sortOrder` (default `""`)
- Validation: `rewriteStrategy` ∈ `{binpack, sort}`; `sortOrder`
required when strategy is `sort`
### How was this patch tested?
```
./gradlew :api:test --tests TestPolicyContents \
:common:test --tests TestPolicyDTO \
:maintenance:optimizer:test \
--tests TestGravitinoCompactionJobAdapter \
--tests TestGravitinoPolicyCompactionStrategy \
-PskipITs -PskipDockerTests=true
```
---------
Co-authored-by: Cursor <[email protected]>
---
.../policy/IcebergDataCompactionContent.java | 85 +++++++++++++++++++++-
.../apache/gravitino/policy/PolicyContents.java | 41 +++++++++++
.../gravitino/policy/TestPolicyContents.java | 53 ++++++++++++++
.../gravitino/dto/policy/PolicyContentDTO.java | 28 +++++++
.../apache/gravitino/dto/util/DTOConverters.java | 4 +
.../apache/gravitino/dto/policy/TestPolicyDTO.java | 6 ++
docs/iceberg-compaction-policy.md | 34 +++++++--
.../handler/compaction/CompactionJobContext.java | 8 ++
.../compaction/CompactionStrategyHandler.java | 22 ++++++
.../job/GravitinoCompactionJobAdapter.java | 13 +++-
.../compaction/TestCompactionStrategyHandler.java | 2 +
.../TestGravitinoPolicyCompactionStrategy.java | 55 ++++++++++++++
.../job/TestBuiltinIcebergRewriteDataFiles.java | 2 +
.../job/TestGravitinoCompactionJobAdapter.java | 19 ++++-
14 files changed, 359 insertions(+), 13 deletions(-)
diff --git
a/api/src/main/java/org/apache/gravitino/policy/IcebergDataCompactionContent.java
b/api/src/main/java/org/apache/gravitino/policy/IcebergDataCompactionContent.java
index 50f76b538f..08c8478d66 100644
---
a/api/src/main/java/org/apache/gravitino/policy/IcebergDataCompactionContent.java
+++
b/api/src/main/java/org/apache/gravitino/policy/IcebergDataCompactionContent.java
@@ -23,6 +23,7 @@ import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import java.util.Collections;
import java.util.LinkedHashMap;
+import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
@@ -42,6 +43,20 @@ public class IcebergDataCompactionContent implements
PolicyContent {
public static final String JOB_TEMPLATE_NAME_VALUE =
"builtin-iceberg-rewrite-data-files";
/** Prefix for rewrite options propagated to job options. */
public static final String JOB_OPTIONS_PREFIX = "job.options.";
+ /**
+ * Rule key for Iceberg {@code rewrite_data_files} top-level {@code
strategy} argument.
+ *
+ * <p>This is separate from {@link #JOB_OPTIONS_PREFIX} because Iceberg
treats {@code strategy} as
+ * a procedure parameter, not an entry in the {@code options} map.
+ */
+ public static final String REWRITE_STRATEGY_KEY = "rewriteStrategy";
+ /**
+ * Rule key for Iceberg {@code rewrite_data_files} top-level {@code
sort_order} argument.
+ *
+ * <p>This is separate from {@link #JOB_OPTIONS_PREFIX} because Iceberg
treats {@code sort_order}
+ * as a procedure parameter, not an entry in the {@code options} map.
+ */
+ public static final String SORT_ORDER_KEY = "sortOrder";
/** Rule key for trigger expression. */
public static final String TRIGGER_EXPR_KEY = "trigger-expr";
/** Rule key for score expression. */
@@ -72,7 +87,17 @@ public class IcebergDataCompactionContent implements
PolicyContent {
public static final long DEFAULT_MAX_PARTITION_NUM = 50L;
/** Default rewrite options for Iceberg rewrite data files. */
public static final Map<String, String> DEFAULT_REWRITE_OPTIONS =
ImmutableMap.of();
+ /** Iceberg rewrite strategy value for binpack. */
+ public static final String REWRITE_STRATEGY_BINPACK = "binpack";
+ /** Iceberg rewrite strategy value for sort. */
+ public static final String REWRITE_STRATEGY_SORT = "sort";
+ /** Default Iceberg rewrite strategy. */
+ public static final String DEFAULT_REWRITE_STRATEGY =
REWRITE_STRATEGY_BINPACK;
+ /** Default Iceberg sort order (empty means unset). */
+ public static final String DEFAULT_SORT_ORDER = "";
+ private static final Set<String> SUPPORTED_REWRITE_STRATEGIES =
+ ImmutableSet.of(REWRITE_STRATEGY_BINPACK, REWRITE_STRATEGY_SORT);
private static final Pattern OPTION_KEY_PATTERN =
Pattern.compile("[A-Za-z0-9._-]+");
private static final Set<MetadataObject.Type> SUPPORTED_OBJECT_TYPES =
ImmutableSet.of(
@@ -99,11 +124,13 @@ public class IcebergDataCompactionContent implements
PolicyContent {
private final Long dataFileMseWeight;
private final Long deleteFileNumberWeight;
private final Long maxPartitionNum;
+ private final String rewriteStrategy;
+ private final String sortOrder;
private final Map<String, String> rewriteOptions;
/** Default constructor for Jackson deserialization only. */
private IcebergDataCompactionContent() {
- this(null, null, null, null, null, null);
+ this(null, null, null, null, null, null, null, null);
}
IcebergDataCompactionContent(
@@ -112,6 +139,8 @@ public class IcebergDataCompactionContent implements
PolicyContent {
Long dataFileMseWeight,
Long deleteFileNumberWeight,
Long maxPartitionNum,
+ String rewriteStrategy,
+ String sortOrder,
Map<String, String> rewriteOptions) {
// Nullable inputs are treated as "use default" to simplify policy
creation.
this.minDataFileMse = minDataFileMse == null ? DEFAULT_MIN_DATA_FILE_MSE :
minDataFileMse;
@@ -122,6 +151,13 @@ public class IcebergDataCompactionContent implements
PolicyContent {
this.deleteFileNumberWeight =
deleteFileNumberWeight == null ? DEFAULT_DELETE_FILE_NUMBER_WEIGHT :
deleteFileNumberWeight;
this.maxPartitionNum = maxPartitionNum == null ? DEFAULT_MAX_PARTITION_NUM
: maxPartitionNum;
+ this.rewriteStrategy =
+ rewriteStrategy == null
+ ? DEFAULT_REWRITE_STRATEGY
+ : rewriteStrategy.trim().toLowerCase(Locale.ROOT);
+ // Null → "" for binpack (Iceberg leaves sort_order unset). When
rewriteStrategy is
+ // "sort", validate() requires a non-blank sortOrder — do not treat "" as
valid for sort.
+ this.sortOrder = sortOrder == null ? DEFAULT_SORT_ORDER : sortOrder.trim();
this.rewriteOptions =
rewriteOptions == null
? DEFAULT_REWRITE_OPTIONS
@@ -173,6 +209,25 @@ public class IcebergDataCompactionContent implements
PolicyContent {
return maxPartitionNum;
}
+ /**
+ * Returns the Iceberg {@code rewrite_data_files} top-level {@code strategy}
argument.
+ *
+ * @return rewrite strategy, one of {@value REWRITE_STRATEGY_BINPACK} or
{@value
+ * REWRITE_STRATEGY_SORT}
+ */
+ public String rewriteStrategy() {
+ return rewriteStrategy;
+ }
+
+ /**
+ * Returns the Iceberg {@code rewrite_data_files} top-level {@code
sort_order} argument.
+ *
+ * @return sort order expression, or empty string when unset
+ */
+ public String sortOrder() {
+ return sortOrder;
+ }
+
/**
* Returns rewrite options that are expanded to {@code job.options.*} rule
entries.
*
@@ -204,6 +259,8 @@ public class IcebergDataCompactionContent implements
PolicyContent {
rules.put(MAX_PARTITION_NUM_KEY, maxPartitionNum);
rules.put(TRIGGER_EXPR_KEY, TRIGGER_EXPR);
rules.put(SCORE_EXPR_KEY, SCORE_EXPR);
+ rules.put(REWRITE_STRATEGY_KEY, rewriteStrategy);
+ rules.put(SORT_ORDER_KEY, sortOrder);
rewriteOptions.forEach((key, value) -> rules.put(JOB_OPTIONS_PREFIX + key,
value));
return Collections.unmodifiableMap(rules);
}
@@ -217,6 +274,22 @@ public class IcebergDataCompactionContent implements
PolicyContent {
Preconditions.checkArgument(dataFileMseWeight >= 0, "dataFileMseWeight
must be >= 0");
Preconditions.checkArgument(deleteFileNumberWeight >= 0,
"deleteFileNumberWeight must be >= 0");
Preconditions.checkArgument(maxPartitionNum > 0, "maxPartitionNum must be
> 0");
+ Preconditions.checkArgument(
+ SUPPORTED_REWRITE_STRATEGIES.contains(rewriteStrategy),
+ "rewriteStrategy must be one of %s, but got '%s'",
+ SUPPORTED_REWRITE_STRATEGIES,
+ rewriteStrategy);
+ if (REWRITE_STRATEGY_SORT.equals(rewriteStrategy)) {
+ Preconditions.checkArgument(
+ StringUtils.isNotBlank(sortOrder),
+ "sortOrder must be set when rewriteStrategy is '%s'",
+ REWRITE_STRATEGY_SORT);
+ } else {
+ Preconditions.checkArgument(
+ StringUtils.isBlank(sortOrder),
+ "sortOrder must be empty when rewriteStrategy is '%s'",
+ rewriteStrategy);
+ }
rewriteOptions.forEach(
(key, value) -> {
@@ -246,6 +319,8 @@ public class IcebergDataCompactionContent implements
PolicyContent {
&& Objects.equals(dataFileMseWeight, that.dataFileMseWeight)
&& Objects.equals(deleteFileNumberWeight, that.deleteFileNumberWeight)
&& Objects.equals(maxPartitionNum, that.maxPartitionNum)
+ && Objects.equals(rewriteStrategy, that.rewriteStrategy)
+ && Objects.equals(sortOrder, that.sortOrder)
&& Objects.equals(rewriteOptions, that.rewriteOptions);
}
@@ -257,6 +332,8 @@ public class IcebergDataCompactionContent implements
PolicyContent {
dataFileMseWeight,
deleteFileNumberWeight,
maxPartitionNum,
+ rewriteStrategy,
+ sortOrder,
rewriteOptions);
}
@@ -273,6 +350,12 @@ public class IcebergDataCompactionContent implements
PolicyContent {
+ deleteFileNumberWeight
+ ", maxPartitionNum="
+ maxPartitionNum
+ + ", rewriteStrategy='"
+ + rewriteStrategy
+ + '\''
+ + ", sortOrder='"
+ + sortOrder
+ + '\''
+ ", rewriteOptions="
+ rewriteOptions
+ '}';
diff --git a/api/src/main/java/org/apache/gravitino/policy/PolicyContents.java
b/api/src/main/java/org/apache/gravitino/policy/PolicyContents.java
index 50e16ac5fe..501475c51f 100644
--- a/api/src/main/java/org/apache/gravitino/policy/PolicyContents.java
+++ b/api/src/main/java/org/apache/gravitino/policy/PolicyContents.java
@@ -55,6 +55,8 @@ public class PolicyContents {
IcebergDataCompactionContent.DEFAULT_DATA_FILE_MSE_WEIGHT,
IcebergDataCompactionContent.DEFAULT_DELETE_FILE_NUMBER_WEIGHT,
IcebergDataCompactionContent.DEFAULT_MAX_PARTITION_NUM,
+ IcebergDataCompactionContent.DEFAULT_REWRITE_STRATEGY,
+ IcebergDataCompactionContent.DEFAULT_SORT_ORDER,
IcebergDataCompactionContent.DEFAULT_REWRITE_OPTIONS);
}
@@ -74,6 +76,8 @@ public class PolicyContents {
IcebergDataCompactionContent.DEFAULT_DATA_FILE_MSE_WEIGHT,
IcebergDataCompactionContent.DEFAULT_DELETE_FILE_NUMBER_WEIGHT,
IcebergDataCompactionContent.DEFAULT_MAX_PARTITION_NUM,
+ IcebergDataCompactionContent.DEFAULT_REWRITE_STRATEGY,
+ IcebergDataCompactionContent.DEFAULT_SORT_ORDER,
rewriteOptions);
}
@@ -95,12 +99,49 @@ public class PolicyContents {
long deleteFileNumberWeight,
long maxPartitionNum,
Map<String, String> rewriteOptions) {
+ return icebergDataCompaction(
+ minDataFileMse,
+ minDeleteFileNumber,
+ dataFileMseWeight,
+ deleteFileNumberWeight,
+ maxPartitionNum,
+ IcebergDataCompactionContent.DEFAULT_REWRITE_STRATEGY,
+ IcebergDataCompactionContent.DEFAULT_SORT_ORDER,
+ rewriteOptions);
+ }
+
+ /**
+ * Creates an iceberg compaction policy content with rewrite strategy and
sort order.
+ *
+ * @param minDataFileMse minimum threshold for custom-data-file-mse
+ * @param minDeleteFileNumber minimum threshold for custom-delete-file-number
+ * @param dataFileMseWeight weight used for custom-data-file-mse score
contribution
+ * @param deleteFileNumberWeight weight used for custom-delete-file-number
score contribution
+ * @param maxPartitionNum maximum partition number selected for compaction
+ * @param rewriteStrategy Iceberg rewrite_data_files top-level strategy
({@code binpack} or {@code
+ * sort})
+ * @param sortOrder Iceberg rewrite_data_files top-level sort_order
(required when strategy is
+ * {@code sort})
+ * @param rewriteOptions rewrite options forwarded as job.options.*
+ * @return iceberg compaction policy content
+ */
+ public static PolicyContent icebergDataCompaction(
+ long minDataFileMse,
+ long minDeleteFileNumber,
+ long dataFileMseWeight,
+ long deleteFileNumberWeight,
+ long maxPartitionNum,
+ String rewriteStrategy,
+ String sortOrder,
+ Map<String, String> rewriteOptions) {
return new IcebergDataCompactionContent(
minDataFileMse,
minDeleteFileNumber,
dataFileMseWeight,
deleteFileNumberWeight,
maxPartitionNum,
+ rewriteStrategy,
+ sortOrder,
rewriteOptions);
}
diff --git
a/api/src/test/java/org/apache/gravitino/policy/TestPolicyContents.java
b/api/src/test/java/org/apache/gravitino/policy/TestPolicyContents.java
index 1112846146..4bac0a9d0f 100644
--- a/api/src/test/java/org/apache/gravitino/policy/TestPolicyContents.java
+++ b/api/src/test/java/org/apache/gravitino/policy/TestPolicyContents.java
@@ -42,6 +42,8 @@ public class TestPolicyContents {
Assertions.assertEquals(1L, content.rules().get("dataFileMseWeight"));
Assertions.assertEquals(100L,
content.rules().get("deleteFileNumberWeight"));
Assertions.assertEquals(50L, content.rules().get("max-partition-num"));
+ Assertions.assertEquals("binpack", content.rules().get("rewriteStrategy"));
+ Assertions.assertEquals("", content.rules().get("sortOrder"));
Assertions.assertNull(content.rules().get("job.options.target-file-size-bytes"));
Assertions.assertNull(content.rules().get("job.options.min-input-files"));
Assertions.assertNull(content.rules().get("job.options.delete-file-threshold"));
@@ -71,6 +73,8 @@ public class TestPolicyContents {
content.rules().get("score-expr"));
Assertions.assertEquals("1048576",
content.rules().get("job.options.target-file-size-bytes"));
Assertions.assertEquals("1",
content.rules().get("job.options.min-input-files"));
+ Assertions.assertEquals("binpack", content.rules().get("rewriteStrategy"));
+ Assertions.assertEquals("", content.rules().get("sortOrder"));
Assertions.assertEquals(
ImmutableSet.of(
MetadataObject.Type.CATALOG, MetadataObject.Type.SCHEMA,
MetadataObject.Type.TABLE),
@@ -95,6 +99,55 @@ public class TestPolicyContents {
Assertions.assertDoesNotThrow(content::validate);
}
+ @Test
+ void testIcebergCompactionContentSupportsSortStrategy() {
+ IcebergDataCompactionContent content =
+ (IcebergDataCompactionContent)
+ PolicyContents.icebergDataCompaction(
+ 1000L, 1L, 1L, 100L, 50L, "sort", "zorder(c1,c2)",
mapOf("min-input-files", "5"));
+
+ Assertions.assertEquals("sort", content.rewriteStrategy());
+ Assertions.assertEquals("zorder(c1,c2)", content.sortOrder());
+ Assertions.assertEquals("sort", content.rules().get("rewriteStrategy"));
+ Assertions.assertEquals("zorder(c1,c2)", content.rules().get("sortOrder"));
+ Assertions.assertEquals("5",
content.rules().get("job.options.min-input-files"));
+ Assertions.assertDoesNotThrow(content::validate);
+ }
+
+ @Test
+ void testIcebergCompactionContentRejectsSortWithoutSortOrder() {
+ IcebergDataCompactionContent content =
+ (IcebergDataCompactionContent)
+ PolicyContents.icebergDataCompaction(1000L, 1L, 1L, 100L, 50L,
"sort", "", Map.of());
+
+ IllegalArgumentException exception =
+ Assertions.assertThrows(IllegalArgumentException.class,
content::validate);
+ Assertions.assertTrue(exception.getMessage().contains("sortOrder"));
+ }
+
+ @Test
+ void testIcebergCompactionContentRejectsBinpackWithSortOrder() {
+ IcebergDataCompactionContent content =
+ (IcebergDataCompactionContent)
+ PolicyContents.icebergDataCompaction(
+ 1000L, 1L, 1L, 100L, 50L, "binpack", "zorder(c1)", Map.of());
+
+ IllegalArgumentException exception =
+ Assertions.assertThrows(IllegalArgumentException.class,
content::validate);
+ Assertions.assertTrue(exception.getMessage().contains("sortOrder must be
empty"));
+ }
+
+ @Test
+ void testIcebergCompactionContentRejectsUnsupportedStrategy() {
+ IcebergDataCompactionContent content =
+ (IcebergDataCompactionContent)
+ PolicyContents.icebergDataCompaction(1000L, 1L, 1L, 100L, 50L,
"unknown", "", Map.of());
+
+ IllegalArgumentException exception =
+ Assertions.assertThrows(IllegalArgumentException.class,
content::validate);
+ Assertions.assertTrue(exception.getMessage().contains("rewriteStrategy"));
+ }
+
@Test
void testIcebergCompactionContentRejectsInvalidRewriteOptionKey() {
IcebergDataCompactionContent content =
diff --git
a/common/src/main/java/org/apache/gravitino/dto/policy/PolicyContentDTO.java
b/common/src/main/java/org/apache/gravitino/dto/policy/PolicyContentDTO.java
index e3f0f77583..65fae911ad 100644
--- a/common/src/main/java/org/apache/gravitino/dto/policy/PolicyContentDTO.java
+++ b/common/src/main/java/org/apache/gravitino/dto/policy/PolicyContentDTO.java
@@ -105,6 +105,12 @@ public interface PolicyContentDTO extends PolicyContent {
@JsonProperty("maxPartitionNum")
private Long maxPartitionNum;
+ @JsonProperty("rewriteStrategy")
+ private String rewriteStrategy;
+
+ @JsonProperty("sortOrder")
+ private String sortOrder;
+
@JsonProperty("rewriteOptions")
private Map<String, String> rewriteOptions;
@@ -166,6 +172,26 @@ public interface PolicyContentDTO extends PolicyContent {
: maxPartitionNum;
}
+ /**
+ * Returns the Iceberg rewrite_data_files top-level strategy.
+ *
+ * @return rewrite strategy
+ */
+ public String rewriteStrategy() {
+ return rewriteStrategy == null
+ ? IcebergDataCompactionContent.DEFAULT_REWRITE_STRATEGY
+ : rewriteStrategy;
+ }
+
+ /**
+ * Returns the Iceberg rewrite_data_files top-level sort_order.
+ *
+ * @return sort order expression
+ */
+ public String sortOrder() {
+ return sortOrder == null ?
IcebergDataCompactionContent.DEFAULT_SORT_ORDER : sortOrder;
+ }
+
/**
* Returns rewrite options expanded to job.options.* during rule
generation.
*
@@ -205,6 +231,8 @@ public interface PolicyContentDTO extends PolicyContent {
dataFileMseWeight(),
deleteFileNumberWeight(),
maxPartitionNum(),
+ rewriteStrategy(),
+ sortOrder(),
rewriteOptions());
}
}
diff --git
a/common/src/main/java/org/apache/gravitino/dto/util/DTOConverters.java
b/common/src/main/java/org/apache/gravitino/dto/util/DTOConverters.java
index 9851e8e4cd..fd45252a6a 100644
--- a/common/src/main/java/org/apache/gravitino/dto/util/DTOConverters.java
+++ b/common/src/main/java/org/apache/gravitino/dto/util/DTOConverters.java
@@ -695,6 +695,8 @@ public class DTOConverters {
.withDataFileMseWeight(icebergCompactionContent.dataFileMseWeight())
.withDeleteFileNumberWeight(icebergCompactionContent.deleteFileNumberWeight())
.withMaxPartitionNum(icebergCompactionContent.maxPartitionNum())
+ .withRewriteStrategy(icebergCompactionContent.rewriteStrategy())
+ .withSortOrder(icebergCompactionContent.sortOrder())
.withRewriteOptions(icebergCompactionContent.rewriteOptions())
.build();
}
@@ -1572,6 +1574,8 @@ public class DTOConverters {
icebergCompactionContentDTO.dataFileMseWeight(),
icebergCompactionContentDTO.deleteFileNumberWeight(),
icebergCompactionContentDTO.maxPartitionNum(),
+ icebergCompactionContentDTO.rewriteStrategy(),
+ icebergCompactionContentDTO.sortOrder(),
icebergCompactionContentDTO.rewriteOptions());
}
diff --git
a/common/src/test/java/org/apache/gravitino/dto/policy/TestPolicyDTO.java
b/common/src/test/java/org/apache/gravitino/dto/policy/TestPolicyDTO.java
index 7bbe897cfc..8dc0292c10 100644
--- a/common/src/test/java/org/apache/gravitino/dto/policy/TestPolicyDTO.java
+++ b/common/src/test/java/org/apache/gravitino/dto/policy/TestPolicyDTO.java
@@ -119,6 +119,8 @@ public class TestPolicyDTO {
.withDataFileMseWeight(2L)
.withDeleteFileNumberWeight(150L)
.withMaxPartitionNum(99L)
+ .withRewriteStrategy("sort")
+ .withSortOrder("zorder(c1,c2)")
.withRewriteOptions(
ImmutableMap.of("target-file-size-bytes", "1048576",
"min-input-files", "1"))
.build();
@@ -169,6 +171,10 @@ public class TestPolicyDTO {
Assertions.assertEquals(
IcebergDataCompactionContent.DEFAULT_MAX_PARTITION_NUM,
contentDTO.maxPartitionNum());
Assertions.assertTrue(contentDTO.rewriteOptions().isEmpty());
+ Assertions.assertEquals(
+ IcebergDataCompactionContent.DEFAULT_REWRITE_STRATEGY,
contentDTO.rewriteStrategy());
+ Assertions.assertEquals(
+ IcebergDataCompactionContent.DEFAULT_SORT_ORDER,
contentDTO.sortOrder());
Assertions.assertDoesNotThrow(contentDTO::validate);
}
}
diff --git a/docs/iceberg-compaction-policy.md
b/docs/iceberg-compaction-policy.md
index 25a5520213..5c78c22621 100644
--- a/docs/iceberg-compaction-policy.md
+++ b/docs/iceberg-compaction-policy.md
@@ -19,14 +19,19 @@ This policy supports `CATALOG`, `SCHEMA`, and `TABLE`
metadata objects.
The typed content for `system_iceberg_compaction` supports the following
fields:
-| Field | Required | Default | Description |
-|---|---|---|---|
-| `minDataFileMse` | No | `405323966463344` | Minimum threshold for metric
`custom-data-file-mse`. Must be `>= 0`. |
-| `minDeleteFileNumber` | No | `1` | Minimum threshold for metric
`custom-delete-file-number`. Must be `>= 0`. |
-| `dataFileMseWeight` | No | `1` | Score weight of `custom-data-file-mse`.
Must be `>= 0`. |
-| `deleteFileNumberWeight` | No | `100` | Score weight of
`custom-delete-file-number`. Must be `>= 0`. |
-| `maxPartitionNum` | No | `50` | Maximum number of partitions selected by
optimizer. Must be `> 0`. |
-| `rewriteOptions` | No | `{}` | Additional rewrite options, expanded as
`job.options.*` rules. |
+| Field | Required | Default | Description
|
+| ------------------------ | -------- | ------------------- |
-------------------------------------------------------------------------------------------------------------
|
+| `minDataFileMse` | No | `405323966463344` | Minimum
threshold for metric `custom-data-file-mse`. Must be `>= 0`.
|
+| `minDeleteFileNumber` | No | `1` | Minimum
threshold for metric `custom-delete-file-number`. Must be `>= 0`.
|
+| `dataFileMseWeight` | No | `1` | Score weight of
`custom-data-file-mse`. Must be `>= 0`.
|
+| `deleteFileNumberWeight` | No | `100` | Score weight of
`custom-delete-file-number`. Must be `>= 0`.
|
+| `maxPartitionNum` | No | `50` | Maximum number
of partitions selected by optimizer. Must be `> 0`.
|
+| `rewriteStrategy` | No | `binpack` | Iceberg
`rewrite_data_files` top-level `strategy`. Supported values: `binpack`, `sort`.
|
+| `sortOrder` | No | `""` | Iceberg
`rewrite_data_files` top-level `sort_order`. Required for `sort`; must be empty
for `binpack`. |
+| `rewriteOptions` | No | `{}` | Iceberg
`rewrite_data_files` `options` map entries, expanded as `job.options.*` rules.
|
+
+`rewriteStrategy` and `sortOrder` are **not** part of `rewriteOptions`.
Iceberg treats them as procedure
+parameters separate from the `options` map.
## Generated Rules and Properties
@@ -39,6 +44,8 @@ The policy content is converted to:
- `trigger-expr=custom-data-file-mse >= minDataFileMse ||
custom-delete-file-number >= minDeleteFileNumber`
- `score-expr=custom-data-file-mse * dataFileMseWeight +
custom-delete-file-number * deleteFileNumberWeight`
- `max-partition-num=<maxPartitionNum>`
+ - `rewriteStrategy=<rewriteStrategy>`
+ - `sortOrder=<sortOrder>`
- `job.options.<key>=<value>` for each rewrite option
## Parameter Tuning Guide
@@ -90,6 +97,17 @@ Recommended `rewriteOptions`:
- `min-input-files = 5`
- `delete-file-threshold = 1`
+### Rewrite Strategy and Sort Order
+
+Use `rewriteStrategy = sort` with a non-empty `sortOrder` when compaction
should rewrite files with
+Iceberg sort or z-order, for example:
+
+- `rewriteStrategy = sort`
+- `sortOrder = zorder(c1,c2)`
+- `sortOrder = id DESC NULLS LAST`
+
+Keep the defaults (`binpack` and empty `sortOrder`) for size-based binpack
compaction.
+
## Policy Examples
<Tabs groupId='language' queryString>
diff --git
a/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/compaction/CompactionJobContext.java
b/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/compaction/CompactionJobContext.java
index 4e36c4f1a3..81d42345f5 100644
---
a/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/compaction/CompactionJobContext.java
+++
b/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/compaction/CompactionJobContext.java
@@ -34,6 +34,8 @@ public class CompactionJobContext implements
JobExecutionContext {
private final NameIdentifier name;
private final Map<String, String> jobOptions;
private final String jobTemplateName;
+ @Getter private final String rewriteStrategy;
+ @Getter private final String sortOrder;
@Getter private final Column[] columns;
@Getter private final Transform[] partitioning;
@Getter private final List<PartitionPath> partitions;
@@ -63,6 +65,12 @@ public class CompactionJobContext implements
JobExecutionContext {
+ ", jobTemplateName='"
+ jobTemplateName
+ '\''
+ + ", rewriteStrategy='"
+ + rewriteStrategy
+ + '\''
+ + ", sortOrder='"
+ + sortOrder
+ + '\''
+ ", jobOptions="
+ jobOptions
+ ", columnCount="
diff --git
a/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/compaction/CompactionStrategyHandler.java
b/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/compaction/CompactionStrategyHandler.java
index 5a5c66d65e..613ebdc393 100644
---
a/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/compaction/CompactionStrategyHandler.java
+++
b/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/compaction/CompactionStrategyHandler.java
@@ -21,6 +21,7 @@ package
org.apache.gravitino.maintenance.optimizer.recommender.handler.compactio
import java.util.EnumSet;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.Set;
import org.apache.gravitino.NameIdentifier;
@@ -63,8 +64,29 @@ public class CompactionStrategyHandler extends
BaseExpressionStrategyHandler {
nameIdentifier,
jobOptions,
strategy.jobTemplateName(),
+ readRewriteStrategy(strategy),
+ readSortOrder(strategy),
tableMetadata.columns(),
tableMetadata.partitioning(),
resolvedPartitions);
}
+
+ private static String readRewriteStrategy(Strategy strategy) {
+ Object value =
strategy.rules().get(IcebergDataCompactionContent.REWRITE_STRATEGY_KEY);
+ if (value == null) {
+ return IcebergDataCompactionContent.DEFAULT_REWRITE_STRATEGY;
+ }
+ String strategyValue = value.toString().trim().toLowerCase(Locale.ROOT);
+ return strategyValue.isEmpty()
+ ? IcebergDataCompactionContent.DEFAULT_REWRITE_STRATEGY
+ : strategyValue;
+ }
+
+ private static String readSortOrder(Strategy strategy) {
+ Object value =
strategy.rules().get(IcebergDataCompactionContent.SORT_ORDER_KEY);
+ if (value == null) {
+ return IcebergDataCompactionContent.DEFAULT_SORT_ORDER;
+ }
+ return value.toString().trim();
+ }
}
diff --git
a/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/job/GravitinoCompactionJobAdapter.java
b/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/job/GravitinoCompactionJobAdapter.java
index 6ca9e67181..8010957dde 100644
---
a/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/job/GravitinoCompactionJobAdapter.java
+++
b/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/job/GravitinoCompactionJobAdapter.java
@@ -29,6 +29,7 @@ import java.util.TreeMap;
import
org.apache.gravitino.maintenance.optimizer.api.recommender.JobExecutionContext;
import org.apache.gravitino.maintenance.optimizer.common.util.IdentifierUtils;
import
org.apache.gravitino.maintenance.optimizer.recommender.handler.compaction.CompactionJobContext;
+import org.apache.gravitino.policy.IcebergDataCompactionContent;
public class GravitinoCompactionJobAdapter implements GravitinoJobAdapter {
@@ -41,11 +42,19 @@ public class GravitinoCompactionJobAdapter implements
GravitinoJobAdapter {
jobExecutionContext instanceof CompactionJobContext,
"jobExecutionContext must be CompactionJobExecutionContext");
CompactionJobContext jobContext = (CompactionJobContext)
jobExecutionContext;
+ String rewriteStrategy =
+ jobContext.getRewriteStrategy() == null
+ ? IcebergDataCompactionContent.DEFAULT_REWRITE_STRATEGY
+ : jobContext.getRewriteStrategy();
+ String sortOrder =
+ jobContext.getSortOrder() == null
+ ? IcebergDataCompactionContent.DEFAULT_SORT_ORDER
+ : jobContext.getSortOrder();
return ImmutableMap.of(
"table_identifier", getTableName(jobContext),
"where_clause", getWhereClause(jobContext),
- "sort_order", "",
- "strategy", "binpack",
+ "sort_order", sortOrder,
+ "strategy", rewriteStrategy,
"options", getOptions(jobContext));
}
diff --git
a/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/compaction/TestCompactionStrategyHandler.java
b/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/compaction/TestCompactionStrategyHandler.java
index 300949a679..2fb96959b8 100644
---
a/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/compaction/TestCompactionStrategyHandler.java
+++
b/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/compaction/TestCompactionStrategyHandler.java
@@ -358,6 +358,8 @@ class TestCompactionStrategyHandler {
tableId,
strategy.jobOptions(),
strategy.jobTemplateName(),
+ "binpack",
+ "",
tableMetadata.columns(),
tableMetadata.partitioning(),
List.of());
diff --git
a/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/compaction/TestGravitinoPolicyCompactionStrategy.java
b/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/compaction/TestGravitinoPolicyCompactionStrategy.java
index f36c28702e..e1e24f4add 100644
---
a/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/compaction/TestGravitinoPolicyCompactionStrategy.java
+++
b/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/compaction/TestGravitinoPolicyCompactionStrategy.java
@@ -26,6 +26,7 @@ import
org.apache.gravitino.maintenance.optimizer.api.common.StatisticEntry;
import
org.apache.gravitino.maintenance.optimizer.api.recommender.StrategyEvaluation;
import
org.apache.gravitino.maintenance.optimizer.api.recommender.StrategyHandlerContext;
import org.apache.gravitino.maintenance.optimizer.common.StatisticEntryImpl;
+import
org.apache.gravitino.maintenance.optimizer.recommender.job.GravitinoCompactionJobAdapter;
import
org.apache.gravitino.maintenance.optimizer.recommender.strategy.GravitinoStrategy;
import org.apache.gravitino.policy.Policy;
import org.apache.gravitino.policy.PolicyContent;
@@ -89,6 +90,60 @@ class TestGravitinoPolicyCompactionStrategy {
Assertions.assertEquals(
Map.of("target-file-size-bytes", "1048576", "min-input-files", "4"),
jobContext.jobOptions());
+ Assertions.assertEquals("binpack", jobContext.getRewriteStrategy());
+ Assertions.assertEquals("", jobContext.getSortOrder());
Assertions.assertTrue(jobContext.getPartitions().isEmpty());
}
+
+ @Test
+ void testPolicyPropagatesSortStrategyToJobContext() {
+ PolicyContent content =
+ PolicyContents.icebergDataCompaction(
+ 1000L,
+ 1L,
+ 2L,
+ 10L,
+ 20L,
+ "sort",
+ "id DESC NULLS LAST",
+ Map.of("target-file-size-bytes", "1048576"));
+ Policy policy = Mockito.mock(Policy.class);
+ Mockito.when(policy.name()).thenReturn("iceberg-compaction-sort-policy");
+ Mockito.when(policy.content()).thenReturn(content);
+
+ GravitinoStrategy strategy = new GravitinoStrategy(policy);
+ Assertions.assertEquals("sort", strategy.rules().get("rewriteStrategy"));
+ Assertions.assertEquals("id DESC NULLS LAST",
strategy.rules().get("sortOrder"));
+ Assertions.assertEquals(Map.of("target-file-size-bytes", "1048576"),
strategy.jobOptions());
+
+ NameIdentifier tableId = NameIdentifier.of("catalog", "db", "table");
+ Table tableMetadata = Mockito.mock(Table.class);
+ Mockito.when(tableMetadata.partitioning()).thenReturn(new Transform[0]);
+ Mockito.when(tableMetadata.columns()).thenReturn(new Column[0]);
+ List<StatisticEntry<?>> tableStatistics =
+ List.of(
+ new StatisticEntryImpl("custom-data-file-mse",
StatisticValues.longValue(3000L)),
+ new StatisticEntryImpl("custom-delete-file-number",
StatisticValues.longValue(5L)));
+
+ StrategyHandlerContext context =
+ StrategyHandlerContext.builder(tableId, strategy)
+ .withTableMetadata(tableMetadata)
+ .withTableStatistics(tableStatistics)
+ .build();
+
+ CompactionStrategyHandler handler = new CompactionStrategyHandler();
+ handler.initialize(context);
+
+ CompactionJobContext jobContext =
+ (CompactionJobContext)
handler.evaluate().jobExecutionContext().orElseThrow();
+ Assertions.assertEquals("sort", jobContext.getRewriteStrategy());
+ Assertions.assertEquals("id DESC NULLS LAST", jobContext.getSortOrder());
+ Assertions.assertEquals(Map.of("target-file-size-bytes", "1048576"),
jobContext.jobOptions());
+
+ GravitinoCompactionJobAdapter adapter = new
GravitinoCompactionJobAdapter();
+ Map<String, String> jobConfig = adapter.jobConfig(jobContext);
+ Assertions.assertEquals("sort", jobConfig.get("strategy"));
+ Assertions.assertEquals("id DESC NULLS LAST", jobConfig.get("sort_order"));
+ Assertions.assertEquals("{\"target-file-size-bytes\":\"1048576\"}",
jobConfig.get("options"));
+ }
}
diff --git
a/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/job/TestBuiltinIcebergRewriteDataFiles.java
b/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/job/TestBuiltinIcebergRewriteDataFiles.java
index 1e49f27214..f1f0a54267 100644
---
a/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/job/TestBuiltinIcebergRewriteDataFiles.java
+++
b/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/job/TestBuiltinIcebergRewriteDataFiles.java
@@ -281,6 +281,8 @@ public class TestBuiltinIcebergRewriteDataFiles {
NameIdentifier.of(SPARK_CATALOG_NAME, "db", tableName),
jobOptions,
JOB_TEMPLATE_NAME,
+ "binpack",
+ "",
columns,
partitioning,
partitions);
diff --git
a/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/job/TestGravitinoCompactionJobAdapter.java
b/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/job/TestGravitinoCompactionJobAdapter.java
index 500138a7fb..dbad1c111b 100644
---
a/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/job/TestGravitinoCompactionJobAdapter.java
+++
b/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/job/TestGravitinoCompactionJobAdapter.java
@@ -40,10 +40,23 @@ public class TestGravitinoCompactionJobAdapter {
"sort_order", "",
"strategy", "binpack",
"options", "{\"target_file_size_bytes\":\"1073741824\"}"),
- jobAdapter.jobConfig(mockCompactionJobContext()));
+ jobAdapter.jobConfig(mockCompactionJobContext("binpack", "")));
}
- private CompactionJobContext mockCompactionJobContext() {
+ @Test
+ public void testSortStrategyAndSortOrder() {
+ GravitinoCompactionJobAdapter jobAdapter = new
GravitinoCompactionJobAdapter();
+ Assertions.assertEquals(
+ Map.of(
+ "table_identifier", "db.table",
+ "where_clause", "",
+ "sort_order", "zorder(c1,c2)",
+ "strategy", "sort",
+ "options", "{\"target_file_size_bytes\":\"1073741824\"}"),
+ jobAdapter.jobConfig(mockCompactionJobContext("sort",
"zorder(c1,c2)")));
+ }
+
+ private CompactionJobContext mockCompactionJobContext(String
rewriteStrategy, String sortOrder) {
String jobTemplateName = "compaction-job-template";
Column[] columns = new Column[0];
Transform[] partitioning = new Transform[0];
@@ -52,6 +65,8 @@ public class TestGravitinoCompactionJobAdapter {
NameIdentifier.of("catalog", "db", "table"),
jobOptions,
jobTemplateName,
+ rewriteStrategy,
+ sortOrder,
columns,
partitioning,
List.of());
