This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 847da32d2bd [fix](iceberg) Make rewrite_manifests rewrite manifests
and count what it did (#69076)
847da32d2bd is described below
commit 847da32d2bde974d5ada37169f2d77d17121d949
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Fri Oct 16 19:14:08 2026 +0800
[fix](iceberg) Make rewrite_manifests rewrite manifests and count what it
did (#69076)
### What problem does this PR solve?
Issue Number: None
Related PR: #59487
Problem Summary:
**In short.** The Iceberg `rewrite_manifests` procedure committed a
snapshot without rewriting any manifest, and still reported every
targeted manifest as rewritten. This PR clusters the data files by runs
of adjacent values of their first partition field, as Trino does, so the
rewrite actually merges manifests with a bounded number of manifest
writers, and returns the numbers of manifests it replaced and wrote. A
unit test written for #68902 turned this up; branch-4.2 has the same
code.
**Background.** `ALTER TABLE ... EXECUTE rewrite_manifests(['spec_id' =
...])` runs `RewriteManifestExecutor` on Iceberg's `RewriteManifests`
API and returns `rewritten_manifests_count` and `added_manifests_count`.
That API merges manifests only when a cluster function is set with
`clusterBy`. Without one, `BaseRewriteManifests.requiresRewrite` returns
false and the commit keeps every current manifest as it is.
**The problem, and what it cost.** The executor never set a cluster
function, so the procedure changed nothing while reporting that it had.
On a local single-BE cluster with a file:// Iceberg hadoop catalog:
| Table | master | with this PR |
|---|---|---|
| Unpartitioned, three INSERTs (3 manifests) | `[3, 0]`, still 3
manifests | `[3, 1]`, 1 manifest |
| Partitioned by `(a, b)`, four INSERTs into `(1, x)`, `(1, y)`, `(2,
x)`, `(2, y)` (4 manifests) | `[4, 0]`, still 4 manifests | `[4, 1]`, 1
manifest: the four fit in one target-size manifest |
| Partitioned by `a`, four INSERTs of `a` = 100, 1, 50, 7,
`commit.manifest.target-size-bytes` set to half the size of the 4
manifests | `[4, 0]`, still 4 manifests | `[4, 2]`, 2 manifests covering
`a` 1 to 7 and 50 to 100 |
| One INSERT (1 small manifest) | `[1, 0]`, plus a new snapshot that
changes nothing | `[0, 0]`, no new snapshot |
Running the procedure to compact metadata never reduced the number of
manifests. Every run on a table with one manifest also added a snapshot.
**How this PR fixes it.**
- **Clustering.** The executor sets `clusterBy` to runs of adjacent
values of the data file's first partition field, as Trino's
`optimize_manifests` does. A partition spec whose targeted manifests are
larger than `commit.manifest.target-size-bytes` (8 MB by default) gets
`min(ceil(their size / target size), 200)` clusters: the executor reads
the distinct values of the spec's first partition field from those
manifests, sorts them with NULL first, and cuts them into that many runs
of about equal length. Adjacent partitions therefore share a manifest,
which keeps each manifest's partition summary narrow, and the rewrite
keeps at most that many manifest writers open for the spec; Iceberg
keeps one open per cluster until the scan ends. A spec whose targeted
manifests fit in one target-size manifest, or an unpartitioned spec, is
one cluster and is not read ahead, so such a table ends up with one
manifest. Iceberg still rolls each cluster over at the target size.
Unlike Trino, which reads every file's partition as the current spec's,
each spec is clustered on its own first field, because `spec_id` can
target an older spec.
- **Single small manifest.** As in Spark's `RewriteManifestsSparkAction`
and in Trino, a single targeted manifest smaller than the target size is
left alone: nothing is committed and the result is `[0, 0]`.
- **Only the targeted manifests.** The rewrite rewrites only the
manifests it targeted when it started, by path, which also applies
`spec_id`. A manifest that another writer commits in between is kept as
it is. Rewriting it failed when it was of a partition spec added after
the table was loaded, because Iceberg's `RewriteManifests` takes the
specs of the table it is created from.
- **Counts.** `rewritten_manifests_count` and `added_manifests_count`
are `manifests-replaced` and `manifests-created` of the snapshot this
rewrite committed, read from the event that Iceberg builds for it after
the commit. The table's current snapshot after the commit can already be
another writer's.
- `spec_id` still limits the rewrite to the manifests of one spec, and
delete manifests are kept as they are, as Iceberg's rewrite does.
Spark's action instead sorts every entry by partition and splits the
entries into `ceil(total size / target size)` manifests. In Spark that
sort is distributed, but here the FE would have to hold every entry in
memory first. This PR keeps Iceberg's streaming cluster function and,
like Trino, holds only the distinct values of the first partition field.
Reading them costs one more pass over the targeted manifests, with only
the partition column projected, and only for a spec whose manifests
exceed the target size.
**Results.**
- **FE UT.** These cases in `IcebergRewriteManifestsActionTest` fail on
master and pass with this PR:
- three appends become one manifest, `[3, 1]`;
- four manifests of a two-field partition that fit in one target-size
manifest become one, `[4, 1]`;
- with the target size set for two clusters, files appended with 100, 1,
50 and 7 become the runs {1, 7} and {50, 100}, `[4, 2]`;
- two partitions that one commit wrote into the same manifest are split
by value, `[2, 2]`;
- a NULL first partition value sorts first and takes the first of two
clusters, `[3, 2]`;
- the manifests of two partition specs, one on an `int` field and one on
a `string` field, are clustered each by its own first field, `[4, 4]`;
- a single small manifest is left alone, `[0, 0]`, with no new snapshot.
On master the clustering cases return `[N, 0]` and the last returns `[1,
0]`. Two cases cover the cluster count (`min(ceil(size / target), 200)`,
without overflow) and the runs (1000 values in 200 clusters, five
adjacent values each, NULL first). Two more cover a writer that commits
after the table was loaded: an appended manifest is kept and the result
is `[2, 1]`, where counting from the table's manifests gave `[2, 2]` and
rewriting every manifest gave `[3, 1]`; and a manifest of a partition
spec added after the load is kept, where the rewrite failed with a
NullPointerException. Another writer promoting the first partition field
from `int` to `long` during the rewrite makes it fail before it commits,
leaving the table as it was. Not sorting, dropping the cap of 200,
rounding up in a way that overflows, clustering every spec on the
current spec's first field, and sorting without NULL first each turn a
case red. All 1540 tests of fe-connector-iceberg pass.
- **End to end.** The table above was produced on a local cluster. After
the rewrite, the four tables still return 3, 4, 1 and 4 rows.
- **Regression.** `test_iceberg_rewrite_manifests` failed on this PR at
its `spec_id` case: the first INSERT after the partition spec changes
merges the three manifests of spec 0 into one, because Iceberg's merge
append compacts the manifests of a spec that the commit does not write
to, and the case rewrote that single small manifest expecting a count of
1. The case now records the manifests per spec in the `.out` (one for
spec 0, two for spec 1), rewrites spec 1 (`[2, 1]`, after which each
spec has one manifest), and checks that rewriting the single small
manifest of spec 0 returns `[0, 0]`. Its other count checks hold with
the new counts. The suite passes on a local cluster against the Iceberg
REST docker environment, where the `.out` was generated.
---
.../iceberg/action/RewriteManifestExecutor.java | 189 +++++++++--
.../connector/iceberg/action/ActionTestTables.java | 4 +
.../action/IcebergRewriteManifestsActionTest.java | 374 ++++++++++++++++++++-
.../action/test_iceberg_rewrite_manifests.out | 14 +
.../action/test_iceberg_rewrite_manifests.groovy | 44 ++-
5 files changed, 559 insertions(+), 66 deletions(-)
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/action/RewriteManifestExecutor.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/action/RewriteManifestExecutor.java
index 31f5b450715..692c58ac85b 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/action/RewriteManifestExecutor.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/action/RewriteManifestExecutor.java
@@ -19,26 +19,57 @@ package org.apache.doris.connector.iceberg.action;
import org.apache.doris.connector.spi.DorisConnectorException;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Iterables;
+import org.apache.iceberg.DataFile;
import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.ManifestReader;
+import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.RewriteManifests;
import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.SnapshotSummary;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.events.CreateSnapshotEvent;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.util.ParallelIterable;
+import org.apache.iceberg.util.PropertyUtil;
+import org.apache.iceberg.util.ThreadPools;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
/**
* Executor for manifest rewrite operations. Connector port of legacy
* {@code datasource/iceberg/rewrite/RewriteManifestExecutor}. The fe-core
couplings are dropped: there is no
* {@code ExternalTable} parameter and no {@code
Env.getExtMetaCacheMgr().invalidateTableCache} call (cache
- * invalidation is performed once at dispatch level by {@code
IcebergProcedureOps}); the SDK call chain and
- * the before/after manifest accounting are otherwise verbatim, with the
failure message kept byte-identical
- * to the legacy {@code UserException}.
+ * invalidation is performed once at dispatch level by {@code
IcebergProcedureOps}); the failure message is kept
+ * byte-identical to the legacy {@code UserException}. Unlike the legacy
executor, which committed a rewrite
+ * without a cluster function (so Iceberg kept every manifest as it was), it
clusters the data files by runs of
+ * adjacent values of their first partition field, skips a single manifest
below the target size, and reports the
+ * manifests the rewrite actually replaced and wrote.
*/
public class RewriteManifestExecutor {
private static final Logger LOG =
LogManager.getLogger(RewriteManifestExecutor.class);
+ /**
+ * The most clusters of one partition spec, and so the most manifest
writers that a rewrite keeps open for it,
+ * as in Trino's optimize_manifests: Iceberg's clusterBy keeps one writer
per key open until the scan ends.
+ */
+ private static final int MAX_CLUSTERS_PER_SPEC = 200;
+
public static class Result {
private final int rewrittenCount;
private final int addedCount;
@@ -66,51 +97,135 @@ public class RewriteManifestExecutor {
}
// Collect manifests before rewrite and filter by specId if
provided
- List<ManifestFile> manifestsBefore =
currentSnapshot.dataManifests(table.io());
- List<ManifestFile> manifestsBeforeTargeted =
filterBySpecId(manifestsBefore, specId);
-
- int rewrittenCount = manifestsBeforeTargeted.size();
-
- if (rewrittenCount == 0) {
+ List<ManifestFile> manifestsBeforeTargeted =
+ filterBySpecId(currentSnapshot.dataManifests(table.io()),
specId);
+ if (manifestsBeforeTargeted.isEmpty()) {
return new Result(0, 0);
}
-
- // Configure rewrite operation, optionally restricting manifests
by specId
- RewriteManifests rm = table.rewriteManifests();
-
- if (specId != null) {
- final int targetSpecId = specId;
- rm.rewriteIf(manifest -> manifest.partitionSpecId() ==
targetSpecId);
+ // One manifest below the target size has nothing to merge with,
as in Spark's and Trino's manifest
+ // rewrites: leave the table alone instead of committing a
snapshot that changes nothing.
+ long targetManifestSizeBytes =
PropertyUtil.propertyAsLong(table.properties(),
+ TableProperties.MANIFEST_TARGET_SIZE_BYTES,
TableProperties.MANIFEST_TARGET_SIZE_BYTES_DEFAULT);
+ if (manifestsBeforeTargeted.size() == 1
+ && manifestsBeforeTargeted.get(0).length() <
targetManifestSizeBytes) {
+ return new Result(0, 0);
}
- // Commit manifest rewrite
+ // Without a cluster function Iceberg rewrites nothing
(BaseRewriteManifests.requiresRewrite) and only
+ // commits a snapshot that keeps every manifest. Only the targeted
manifests are rewritten: a manifest
+ // that another writer commits before the rewrite applies is kept
as it is. Such a manifest can be of a
+ // partition spec added after the table was loaded, which the
rewrite cannot write, because it takes
+ // the specs of the table it is created from.
+ Set<String> targetedPaths = manifestsBeforeTargeted.stream()
+ .map(ManifestFile::path)
+ .collect(Collectors.toSet());
+ Map<Integer, Map<Object, Integer>> clustersBySpec =
+ clustersBySpec(table, manifestsBeforeTargeted,
targetManifestSizeBytes);
+ RewriteManifests rm = table.rewriteManifests()
+ .clusterBy(file -> clusterOf(clustersBySpec, file))
+ .rewriteIf(manifest ->
targetedPaths.contains(manifest.path()));
rm.commit();
- // Refresh snapshot after rewrite
- Snapshot snapshotAfter = table.currentSnapshot();
- if (snapshotAfter == null) {
- return new Result(rewrittenCount, 0);
+ // Count the manifests replaced and written by the snapshot this
rewrite committed. Iceberg builds the
+ // event of that snapshot after the commit
(SnapshotProducer.notifyListeners), while the table's current
+ // snapshot can already be another one.
+ Map<String, String> summary = ((CreateSnapshotEvent)
rm.updateEvent()).summary();
+ return new
Result(Integer.parseInt(summary.get(SnapshotSummary.REPLACED_MANIFESTS_COUNT)),
+
Integer.parseInt(summary.get(SnapshotSummary.CREATED_MANIFESTS_COUNT)));
+ } catch (Exception e) {
+ LOG.warn("Failed to execute manifest rewrite for table: {}",
table.name(), e);
+ throw new DorisConnectorException("Failed to rewrite manifests: "
+ e.getMessage(), e);
+ }
+ }
+
+ /**
+ * Groups the data files of each targeted partition spec into runs of
adjacent values of the spec's first
+ * partition field, as Trino's optimize_manifests does. A spec gets
min(ceil(size of its targeted manifests /
+ * target size), 200) clusters: the distinct values are read from its
targeted manifests, sorted with NULL first,
+ * and the value at rank i goes to cluster i * clusters / n. Adjacent
partitions then share a manifest, which
+ * keeps the manifests' partition summaries narrow for pruning, while the
rewrite keeps at most that many
+ * writers open for the spec; Iceberg still rolls each cluster over at the
target size. An unpartitioned spec, or
+ * one whose targeted manifests fit in one target-size manifest, is one
cluster and is not read ahead. Each spec
+ * is clustered on its own first field, which can be of another type than
another spec's: Iceberg keeps a writer
+ * per cluster and spec, so the specs' cluster numbers do not collide.
+ *
+ * @return the value-to-cluster map of each targeted spec, empty for a
spec that is one cluster
+ */
+ private static Map<Integer, Map<Object, Integer>> clustersBySpec(Table
table, List<ManifestFile> targeted,
+ long targetManifestSizeBytes) throws IOException {
+ Map<Integer, List<ManifestFile>> manifestsBySpec = targeted.stream()
+ .collect(Collectors.groupingBy(ManifestFile::partitionSpecId));
+ Map<Integer, Map<Object, Integer>> clustersBySpec = new HashMap<>();
+ for (Map.Entry<Integer, List<ManifestFile>> entry :
manifestsBySpec.entrySet()) {
+ PartitionSpec spec = table.specs().get(entry.getKey());
+ int clusters =
clusterCount(entry.getValue().stream().mapToLong(ManifestFile::length).sum(),
+ targetManifestSizeBytes);
+ if (spec.isUnpartitioned() || clusters == 1) {
+ clustersBySpec.put(entry.getKey(), Collections.emptyMap());
+ continue;
}
+ Comparator<Object> order =
Comparators.<Object>nullsFirst().thenComparing(
+
Comparators.forType(spec.partitionType().fields().get(0).type().asPrimitiveType()));
+ clustersBySpec.put(entry.getKey(),
+ assignClusters(firstPartitionValues(table,
entry.getValue()), order, clusters));
+ }
+ return clustersBySpec;
+ }
- // Collect manifests after rewrite and filter by specId
- List<ManifestFile> manifestsAfter =
snapshotAfter.dataManifests(table.io());
- List<ManifestFile> manifestsAfterTargeted =
filterBySpecId(manifestsAfter, specId);
+ /** min(ceil(manifestsSizeBytes / targetManifestSizeBytes), 200); a
manifest is never empty, so at least 1. */
+ static int clusterCount(long manifestsSizeBytes, long
targetManifestSizeBytes) {
+ return (int) Math.min((manifestsSizeBytes - 1) /
targetManifestSizeBytes + 1, MAX_CLUSTERS_PER_SPEC);
+ }
- // Compute addedCount as newly produced manifests (path not in
before set)
- java.util.Set<String> beforePaths =
manifestsBeforeTargeted.stream()
- .map(ManifestFile::path)
- .collect(java.util.stream.Collectors.toSet());
+ /**
+ * Sorts the values and gives the one at rank i cluster i * clusters / n,
so each cluster is a run of adjacent
+ * values and the runs differ in length by at most one.
+ */
+ static Map<Object, Integer> assignClusters(Set<Object> values,
Comparator<Object> order, int clusters) {
+ List<Object> sorted = new ArrayList<>(values);
+ sorted.sort(order);
+ Map<Object, Integer> clusterOfValue = new HashMap<>();
+ for (int rank = 0; rank < sorted.size(); rank++) {
+ clusterOfValue.put(sorted.get(rank), (int) ((long) rank * clusters
/ sorted.size()));
+ }
+ return Collections.unmodifiableMap(clusterOfValue);
+ }
- int addedCount = (int) manifestsAfterTargeted.stream()
- .map(ManifestFile::path)
- .filter(path -> !beforePaths.contains(path))
- .count();
+ /**
+ * The distinct values of the first partition field among the live data
files of manifests of one spec, read
+ * with only the partition column projected, in parallel on Iceberg's
worker pool as the rewrite itself reads
+ * them. ManifestReader returns copies of the files, and PartitionData
returns immutable values or copies of
+ * binary ones, so the values can be kept and compared by content; a
HashSet holds a NULL value as well.
+ */
+ private static Set<Object> firstPartitionValues(Table table,
List<ManifestFile> manifests) throws IOException {
+ Iterable<ManifestReader<DataFile>> readers =
Iterables.transform(manifests,
+ manifest -> ManifestFiles.read(manifest, table.io(),
table.specs())
+ .select(Collections.singletonList("partition")));
+ Set<Object> values = new HashSet<>();
+ try (CloseableIterable<DataFile> files = new
ParallelIterable<>(readers, ThreadPools.getWorkerPool())) {
+ for (DataFile file : files) {
+ values.add(file.partition().get(0, Object.class));
+ }
+ }
+ return values;
+ }
- return new Result(rewrittenCount, addedCount);
- } catch (Exception e) {
- LOG.warn("Failed to execute manifest rewrite for table: {}",
table.name(), e);
- throw new DorisConnectorException("Failed to rewrite manifests: "
+ e.getMessage(), e);
+ /**
+ * The cluster of a data file. Iceberg calls this from its worker threads
while it reads the targeted manifests,
+ * so it only reads maps built before the commit. Every file it is given
comes from a targeted manifest, whose
+ * values were read ahead; a value missing from the map means the table's
partitioning changed while the
+ * rewrite ran.
+ */
+ private static Object clusterOf(Map<Integer, Map<Object, Integer>>
clustersBySpec, DataFile file) {
+ Map<Object, Integer> clusters = clustersBySpec.get(file.specId());
+ if (clusters.isEmpty()) {
+ return 0;
}
+ Object value = file.partition().get(0, Object.class);
+ Integer cluster = clusters.get(value);
+ Preconditions.checkState(cluster != null, "Partition value %s of %s
was not in the targeted manifests when"
+ + " the rewrite started; the table's partitioning changed
during the rewrite", value, file.location());
+ return cluster;
}
private List<ManifestFile> filterBySpecId(List<ManifestFile> manifests,
Integer specId) {
@@ -120,6 +235,6 @@ public class RewriteManifestExecutor {
final int targetSpecId = specId;
return manifests.stream()
.filter(manifest -> manifest.partitionSpecId() == targetSpecId)
- .collect(java.util.stream.Collectors.toList());
+ .collect(Collectors.toList());
}
}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/action/ActionTestTables.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/action/ActionTestTables.java
index 49ffde1fb47..f0f739b62e2 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/action/ActionTestTables.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/action/ActionTestTables.java
@@ -69,6 +69,10 @@ final class ActionTestTables {
return createTable(catalog, table, Collections.emptyMap());
}
+ static Table createTable(InMemoryCatalog catalog, String table,
PartitionSpec spec) {
+ return catalog.createTable(id(table), SCHEMA, spec,
Collections.emptyMap());
+ }
+
/** Appends one data file as a new snapshot and returns the new current
snapshot id. */
static long appendSnapshot(InMemoryCatalog catalog, String table, String
fileName, long records) {
Table t = catalog.loadTable(id(table));
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/action/IcebergRewriteManifestsActionTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/action/IcebergRewriteManifestsActionTest.java
index 4f538c4ae1e..7e6b4fe0f9e 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/action/IcebergRewriteManifestsActionTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/action/IcebergRewriteManifestsActionTest.java
@@ -22,12 +22,19 @@ import
org.apache.doris.connector.spi.procedure.ConnectorProcedureResult;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Iterables;
import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
import org.apache.iceberg.DeleteFiles;
import org.apache.iceberg.ExpireSnapshots;
+import org.apache.iceberg.FileFormat;
import org.apache.iceberg.HistoryEntry;
import org.apache.iceberg.ManageSnapshots;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestFiles;
import org.apache.iceberg.OverwriteFiles;
+import org.apache.iceberg.PartitionData;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.ReplacePartitions;
import org.apache.iceberg.ReplaceSortOrder;
@@ -40,6 +47,7 @@ import org.apache.iceberg.SnapshotRef;
import org.apache.iceberg.SortOrder;
import org.apache.iceberg.StatisticsFile;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
import org.apache.iceberg.TableScan;
import org.apache.iceberg.Transaction;
import org.apache.iceberg.UpdateLocation;
@@ -49,14 +57,23 @@ import org.apache.iceberg.UpdateSchema;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.encryption.EncryptionManager;
import org.apache.iceberg.inmemory.InMemoryCatalog;
+import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.io.LocationProvider;
+import org.apache.iceberg.types.Comparators;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
/**
* Pins {@code rewrite_manifests}, including the delegated connector {@link
RewriteManifestExecutor}.
@@ -103,12 +120,357 @@ public class IcebergRewriteManifestsActionTest {
action.validate();
ConnectorProcedureResult result =
action.execute(catalog.loadTable(id), ActionTestTables.session("UTC"));
- // The executor reports manifestsBefore.size() as the rewritten count
(the faithful port behaviour);
- // the added count is iceberg's internal combine outcome, asserted
only as a valid non-negative int.
- Assertions.assertEquals(String.valueOf(manifestsBefore),
result.getRows().get(0).get(0),
- "rewritten_manifests_count is the number of manifests targeted
for rewrite");
- Assertions.assertTrue(Integer.parseInt(result.getRows().get(0).get(1))
>= 0,
- "added_manifests_count is a valid non-negative count");
+ // Iceberg rewrites nothing without a cluster function
(BaseRewriteManifests.requiresRewrite) and only
+ // commits a snapshot that keeps every manifest. With one, the three
manifests of an unpartitioned table
+ // become one, and the counts are the manifests actually replaced and
written.
+ Assertions.assertEquals(ImmutableList.of("3", "1"),
result.getRows().get(0));
+ Table after = catalog.loadTable(id);
+ Assertions.assertEquals(1,
after.currentSnapshot().dataManifests(after.io()).size());
+ Assertions.assertEquals(3,
Iterables.size(after.newScan().planFiles()), "every data file is kept");
+ }
+
+ @Test
+ public void mergesPartitionedManifestsThatFitTheTargetSizeIntoOne() {
+ // A partition spec gets min(ceil(size of its manifests / target
size), 200) clusters, as in Trino's
+ // optimize_manifests. Four small manifests fit in one target-size
manifest, so they become one.
+ InMemoryCatalog catalog = ActionTestTables.freshCatalog();
+ PartitionSpec spec =
PartitionSpec.builderFor(ActionTestTables.SCHEMA).identity("id").identity("name")
+ .build();
+ ActionTestTables.createTable(catalog, "t", spec);
+ append(catalog, partitionedFile(spec, "f1.parquet", 1, "a"));
+ append(catalog, partitionedFile(spec, "f2.parquet", 1, "b"));
+ append(catalog, partitionedFile(spec, "f3.parquet", 2, "a"));
+ append(catalog, partitionedFile(spec, "f4.parquet", 2, "b"));
+
+ ConnectorProcedureResult result = rewriteAll(catalog);
+
+ Assertions.assertEquals(ImmutableList.of("4", "1"),
result.getRows().get(0));
+ Assertions.assertEquals(Collections.singletonList(new
HashSet<>(Arrays.asList(1, 2))),
+ firstPartitionValuesPerManifest(catalog));
+ }
+
+ @Test
+ public void clustersAdjacentFirstPartitionValues() {
+ // With two clusters, the sorted values 1, 7, 50 and 100 go to runs
{1, 7} and {50, 100}, so each manifest
+ // covers a narrow range of the first partition field. The files are
appended out of order, and a HashSet
+ // of these Integers iterates 1, 50, 100, 7, so clustering the values
unsorted would mix the runs.
+ // MUTATION: not sorting the values -> red.
+ InMemoryCatalog catalog = ActionTestTables.freshCatalog();
+ PartitionSpec spec =
PartitionSpec.builderFor(ActionTestTables.SCHEMA).identity("id").identity("name")
+ .build();
+ ActionTestTables.createTable(catalog, "t", spec);
+ append(catalog, partitionedFile(spec, "f1.parquet", 100, "a"));
+ append(catalog, partitionedFile(spec, "f2.parquet", 1, "b"));
+ append(catalog, partitionedFile(spec, "f3.parquet", 50, "a"));
+ append(catalog, partitionedFile(spec, "f4.parquet", 7, "b"));
+ targetClusters(catalog, 2);
+
+ ConnectorProcedureResult result = rewriteAll(catalog);
+
+ Assertions.assertEquals(ImmutableList.of("4", "2"),
result.getRows().get(0));
+ Assertions.assertEquals(
+ Arrays.asList(new HashSet<>(Arrays.asList(1, 7)), new
HashSet<>(Arrays.asList(50, 100))),
+ firstPartitionValuesPerManifest(catalog));
+ }
+
+ @Test
+ public void clustersPartitionsReadFromTheSameManifest() {
+ // One commit can write several partitions into one manifest: each
commit below writes both, and with two
+ // clusters the rewrite splits them by partition value.
+ InMemoryCatalog catalog = ActionTestTables.freshCatalog();
+ PartitionSpec spec =
PartitionSpec.builderFor(ActionTestTables.SCHEMA).identity("id").build();
+ ActionTestTables.createTable(catalog, "t", spec);
+ append(catalog, partitionedFile(spec, "f1.parquet", 1),
partitionedFile(spec, "f2.parquet", 2));
+ append(catalog, partitionedFile(spec, "f3.parquet", 1),
partitionedFile(spec, "f4.parquet", 2));
+ targetClusters(catalog, 2);
+
+ ConnectorProcedureResult result = rewriteAll(catalog);
+
+ Assertions.assertEquals(ImmutableList.of("2", "2"),
result.getRows().get(0));
+ assertOneFirstPartitionValuePerManifest(catalog, Arrays.asList(1, 2));
+ }
+
+ @Test
+ public void clustersANullFirstPartitionValue() {
+ // A NULL partition value sorts first, so with two clusters it takes
the first and 1 the second. Iceberg
+ // rejects a null cluster key, and the cluster of a NULL value is a
number like any other.
+ // MUTATION: sorting without NULLs first -> red (the comparator does
not take a null).
+ InMemoryCatalog catalog = ActionTestTables.freshCatalog();
+ PartitionSpec spec =
PartitionSpec.builderFor(ActionTestTables.SCHEMA).identity("id").identity("name")
+ .build();
+ ActionTestTables.createTable(catalog, "t", spec);
+ append(catalog, partitionedFile(spec, "f1.parquet", null, "a"));
+ append(catalog, partitionedFile(spec, "f2.parquet", null, "b"));
+ append(catalog, partitionedFile(spec, "f3.parquet", 1, "a"));
+ targetClusters(catalog, 2);
+
+ ConnectorProcedureResult result = rewriteAll(catalog);
+
+ Assertions.assertEquals(ImmutableList.of("3", "2"),
result.getRows().get(0));
+ assertOneFirstPartitionValuePerManifest(catalog, Arrays.asList(null,
1));
+ }
+
+ @Test
+ public void clustersEachPartitionSpecByItsOwnFirstField() {
+ // After the spec changed from identity(id) to identity(name), the
manifests of each spec are clustered by
+ // that spec's own first field, an int in one and a string in the
other: one comparator for all the values
+ // would fail on the mix. The files are fast-appended, since a merge
append would also merge the manifests
+ // of the old spec. MUTATION: clustering every spec with the current
spec's first field -> red.
+ InMemoryCatalog catalog = ActionTestTables.freshCatalog();
+ TableIdentifier id = ActionTestTables.id("t");
+ PartitionSpec idSpec =
PartitionSpec.builderFor(ActionTestTables.SCHEMA).identity("id").build();
+ ActionTestTables.createTable(catalog, "t", idSpec);
+
catalog.loadTable(id).newFastAppend().appendFile(partitionedFile(idSpec,
"f1.parquet", 1)).commit();
+
catalog.loadTable(id).newFastAppend().appendFile(partitionedFile(idSpec,
"f2.parquet", 2)).commit();
+
catalog.loadTable(id).updateSpec().removeField("id").addField("name").commit();
+ PartitionSpec nameSpec = catalog.loadTable(id).spec();
+
catalog.loadTable(id).newFastAppend().appendFile(partitionedFile(nameSpec,
"f3.parquet", "a")).commit();
+
catalog.loadTable(id).newFastAppend().appendFile(partitionedFile(nameSpec,
"f4.parquet", "b")).commit();
+ targetClusters(catalog, 4);
+
+ ConnectorProcedureResult result = rewriteAll(catalog);
+
+ Assertions.assertEquals(ImmutableList.of("4", "4"),
result.getRows().get(0));
+ assertOneFirstPartitionValuePerManifest(catalog, Arrays.asList(1, 2,
"a", "b"));
+ }
+
+ @Test
+ public void
failsBeforeCommitWhenTheFirstPartitionFieldIsPromotedDuringTheRewrite() {
+ // Another writer promotes the identity source column from INT to LONG
after the table was loaded. Iceberg
+ // rebuilds the existing specs with the new type under the same ids,
so the rewrite reads the targeted
+ // manifests' values as Longs while the values read ahead were
Integers. The rewrite fails before it
+ // commits, and the table is left as it was.
+ InMemoryCatalog catalog = ActionTestTables.freshCatalog();
+ TableIdentifier id = ActionTestTables.id("t");
+ PartitionSpec spec =
PartitionSpec.builderFor(ActionTestTables.SCHEMA).identity("id").build();
+ ActionTestTables.createTable(catalog, "t", spec);
+ catalog.loadTable(id).newFastAppend().appendFile(partitionedFile(spec,
"f1.parquet", 1)).commit();
+ catalog.loadTable(id).newFastAppend().appendFile(partitionedFile(spec,
"f2.parquet", 2)).commit();
+ targetClusters(catalog, 2);
+ Table loaded = catalog.loadTable(id);
+ catalog.loadTable(id).updateSchema().updateColumn("id",
org.apache.iceberg.types.Types.LongType.get())
+ .commit();
+ Set<String> manifestsBefore = manifestPaths(catalog.loadTable(id));
+
+ DorisConnectorException e =
Assertions.assertThrows(DorisConnectorException.class, () -> rewrite(loaded));
+
+ Assertions.assertTrue(e.getMessage().contains("the table's
partitioning changed during the rewrite"),
+ e.getMessage());
+ Table after = catalog.loadTable(id);
+ Assertions.assertEquals(loaded.currentSnapshot().snapshotId(),
after.currentSnapshot().snapshotId());
+ Assertions.assertEquals(manifestsBefore, manifestPaths(after));
+ }
+
+ @Test
+ public void clusterCountIsTheManifestSizeOverTheTargetUpTo200() {
+ // As in Trino, at most 200 clusters, and so at most 200 open manifest
writers, per spec; the division
+ // rounds up without overflowing. MUTATION: no cap, or rounding up as
(size + target - 1) / target -> red.
+ long target = 8L * 1024 * 1024;
+ Assertions.assertEquals(1, RewriteManifestExecutor.clusterCount(1,
target));
+ Assertions.assertEquals(1,
RewriteManifestExecutor.clusterCount(target, target));
+ Assertions.assertEquals(2, RewriteManifestExecutor.clusterCount(target
+ 1, target));
+ Assertions.assertEquals(200, RewriteManifestExecutor.clusterCount(200
* target, target));
+ Assertions.assertEquals(200,
RewriteManifestExecutor.clusterCount(Long.MAX_VALUE, target));
+ Assertions.assertEquals(3,
RewriteManifestExecutor.clusterCount(Long.MAX_VALUE, Long.MAX_VALUE / 2));
+ }
+
+ @Test
+ public void assignsSortedValuesToRunsOfNearlyEqualLength() {
+ // 1000 values in 200 clusters: five adjacent values per cluster, NULL
first. Three values in 200 clusters
+ // are three clusters.
+ Comparator<Object> order =
Comparators.<Object>nullsFirst().thenComparing(
+
Comparators.forType(org.apache.iceberg.types.Types.IntegerType.get()));
+ Set<Object> values = new HashSet<>();
+ values.add(null);
+ for (int i = 0; i < 999; i++) {
+ values.add(i);
+ }
+
+ Map<Object, Integer> clusters =
RewriteManifestExecutor.assignClusters(values, order, 200);
+
+ Assertions.assertEquals(0, clusters.get(null).intValue());
+ int[] sizes = new int[200];
+ int previous = clusters.get(null);
+ sizes[previous]++;
+ for (int i = 0; i < 999; i++) {
+ int cluster = clusters.get(i);
+ Assertions.assertTrue(cluster >= previous, "cluster numbers follow
the sort order");
+ sizes[cluster]++;
+ previous = cluster;
+ }
+ for (int size : sizes) {
+ Assertions.assertEquals(5, size);
+ }
+ Assertions.assertEquals(3, new
HashSet<>(RewriteManifestExecutor.assignClusters(
+ new HashSet<>(Arrays.asList(3, 1, 2)), order,
200).values()).size());
+ }
+
+ @Test
+ public void leavesASingleSmallManifestUntouched() {
+ // One manifest below commit.manifest.target-size-bytes has nothing to
merge with, as in Spark's and
+ // Trino's manifest rewrites: nothing is committed and nothing is
counted.
+ InMemoryCatalog catalog = ActionTestTables.freshCatalog();
+ TableIdentifier id = ActionTestTables.id("t");
+ ActionTestTables.createTable(catalog, "t");
+ ActionTestTables.appendSnapshot(catalog, "t", "f1.parquet", 1L);
+ long snapshotBefore =
catalog.loadTable(id).currentSnapshot().snapshotId();
+
+ ConnectorProcedureResult result = rewriteAll(catalog);
+
+ Assertions.assertEquals(ImmutableList.of("0", "0"),
result.getRows().get(0));
+ Assertions.assertEquals(snapshotBefore,
catalog.loadTable(id).currentSnapshot().snapshotId(),
+ "no snapshot is committed");
+ }
+
+ @Test
+ public void keepsAManifestCommittedAfterTheTableWasLoaded() {
+ // The rewrite targets the manifests of the snapshot the table was
loaded with. A manifest that another
+ // writer commits before the rewrite applies is kept as it is, and the
counts are those of the snapshot
+ // the rewrite committed, which the table's manifests before and after
it would not give.
+ InMemoryCatalog catalog = ActionTestTables.freshCatalog();
+ TableIdentifier id = ActionTestTables.id("t");
+ ActionTestTables.createTable(catalog, "t");
+ ActionTestTables.appendSnapshot(catalog, "t", "f1.parquet", 1L);
+ ActionTestTables.appendSnapshot(catalog, "t", "f2.parquet", 2L);
+ Table loaded = catalog.loadTable(id);
+ ActionTestTables.appendSnapshot(catalog, "t", "f3.parquet", 3L);
+ Set<String> concurrentManifests = manifestPaths(catalog.loadTable(id));
+ concurrentManifests.removeAll(manifestPaths(loaded));
+ Assertions.assertEquals(1, concurrentManifests.size());
+
+ ConnectorProcedureResult result = rewrite(loaded);
+
+ Assertions.assertEquals(ImmutableList.of("2", "1"),
result.getRows().get(0));
+ Table after = catalog.loadTable(id);
+ Set<String> manifestsAfter = manifestPaths(after);
+ Assertions.assertEquals(2, manifestsAfter.size(),
manifestsAfter.toString());
+ Assertions.assertTrue(manifestsAfter.containsAll(concurrentManifests),
manifestsAfter.toString());
+ Assertions.assertEquals(3,
Iterables.size(after.newScan().planFiles()), "every data file is kept");
+ }
+
+ @Test
+ public void keepsAManifestOfAPartitionSpecAddedAfterTheTableWasLoaded() {
+ // Iceberg's RewriteManifests takes the partition specs of the table
it is created from, so it cannot
+ // rewrite a manifest of a spec that another writer adds after the
table was loaded: a rewrite that
+ // reached it failed with a NullPointerException. That manifest is
kept as it is. The other writer fast
+ // appends, because a merge append would also merge the manifests of
the old spec.
+ InMemoryCatalog catalog = ActionTestTables.freshCatalog();
+ TableIdentifier id = ActionTestTables.id("t");
+ ActionTestTables.createTable(catalog, "t");
+ ActionTestTables.appendSnapshot(catalog, "t", "f1.parquet", 1L);
+ ActionTestTables.appendSnapshot(catalog, "t", "f2.parquet", 2L);
+ Table loaded = catalog.loadTable(id);
+ catalog.loadTable(id).updateSpec().addField("id").commit();
+ Table evolved = catalog.loadTable(id);
+ evolved.newFastAppend().appendFile(partitionedFile(evolved.spec(),
"f3.parquet", 3)).commit();
+
+ ConnectorProcedureResult result = rewrite(loaded);
+
+ Assertions.assertEquals(ImmutableList.of("2", "1"),
result.getRows().get(0));
+ Table after = catalog.loadTable(id);
+ List<Integer> specIds = new ArrayList<>();
+ for (ManifestFile manifest :
after.currentSnapshot().dataManifests(after.io())) {
+ specIds.add(manifest.partitionSpecId());
+ }
+ Collections.sort(specIds);
+ Assertions.assertEquals(Arrays.asList(0, 1), specIds);
+ Assertions.assertEquals(3,
Iterables.size(after.newScan().planFiles()), "every data file is kept");
+ }
+
+ private static ConnectorProcedureResult rewriteAll(InMemoryCatalog
catalog) {
+ return rewrite(catalog.loadTable(ActionTestTables.id("t")));
+ }
+
+ private static ConnectorProcedureResult rewrite(Table table) {
+ IcebergRewriteManifestsAction action = action(Collections.emptyMap());
+ action.validate();
+ return action.execute(table, ActionTestTables.session("UTC"));
+ }
+
+ private static Set<String> manifestPaths(Table table) {
+ Set<String> paths = new HashSet<>();
+ for (ManifestFile manifest :
table.currentSnapshot().dataManifests(table.io())) {
+ paths.add(manifest.path());
+ }
+ return paths;
+ }
+
+ private static DataFile partitionedFile(PartitionSpec spec, String
fileName, Object... partitionValues) {
+ PartitionData partition = new PartitionData(spec.partitionType());
+ for (int i = 0; i < partitionValues.length; i++) {
+ partition.set(i, partitionValues[i]);
+ }
+ return DataFiles.builder(spec)
+ .withPath("s3://b/db1/" + fileName)
+ .withFileSizeInBytes(1024)
+ .withRecordCount(1)
+ .withFormat(FileFormat.PARQUET)
+ .withPartition(partition)
+ .build();
+ }
+
+ private static void append(InMemoryCatalog catalog, DataFile... files) {
+ AppendFiles append =
catalog.loadTable(ActionTestTables.id("t")).newAppend();
+ for (DataFile file : files) {
+ append.appendFile(file);
+ }
+ append.commit();
+ }
+
+ /**
+ * Sets the target manifest size so that t's data manifests make the given
number of clusters: each manifest
+ * is a few KB, so ceil(total / ceil(total / clusters)) is clusters, and a
rewritten manifest, which holds one
+ * header for the files of several, stays below the target.
+ */
+ private static void targetClusters(InMemoryCatalog catalog, int clusters) {
+ Table table = catalog.loadTable(ActionTestTables.id("t"));
+ long total = 0;
+ for (ManifestFile manifest :
table.currentSnapshot().dataManifests(table.io())) {
+ total += manifest.length();
+ }
+ table.updateProperties()
+ .set(TableProperties.MANIFEST_TARGET_SIZE_BYTES,
String.valueOf((total + clusters - 1) / clusters))
+ .commit();
+ }
+
+ /** The first-partition-field values of each data manifest of t, ordered
by the manifests' smallest values. */
+ private static List<Set<Object>>
firstPartitionValuesPerManifest(InMemoryCatalog catalog) {
+ Table table = catalog.loadTable(ActionTestTables.id("t"));
+ List<Set<Object>> manifests = new ArrayList<>();
+ for (ManifestFile manifest :
table.currentSnapshot().dataManifests(table.io())) {
+ Set<Object> manifestValues = new HashSet<>();
+ try (CloseableIterable<DataFile> files =
ManifestFiles.read(manifest, table.io(), table.specs())) {
+ for (DataFile file : files) {
+ manifestValues.add(file.partition().get(0, Object.class));
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ manifests.add(manifestValues);
+ }
+ manifests.sort(Comparator.comparingInt((Set<Object> values) ->
values.stream()
+ .mapToInt(value -> (Integer) value).min().getAsInt()));
+ return manifests;
+ }
+
+ /** Every data manifest holds files of one first-partition-field value,
and together they hold expected. */
+ private static void
assertOneFirstPartitionValuePerManifest(InMemoryCatalog catalog, List<?>
expected) {
+ Table table = catalog.loadTable(ActionTestTables.id("t"));
+ List<Object> values = new ArrayList<>();
+ for (ManifestFile manifest :
table.currentSnapshot().dataManifests(table.io())) {
+ Set<Object> manifestValues = new HashSet<>();
+ try (CloseableIterable<DataFile> files =
ManifestFiles.read(manifest, table.io(), table.specs())) {
+ for (DataFile file : files) {
+ manifestValues.add(file.partition().get(0, Object.class));
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ Assertions.assertEquals(1, manifestValues.size(), manifest.path()
+ " holds " + manifestValues);
+ values.addAll(manifestValues);
+ }
+ Assertions.assertEquals(new HashSet<>(expected), new
HashSet<>(values));
+ Assertions.assertEquals(expected.size(), values.size(), "one manifest
per first partition value");
}
@Test
diff --git
a/regression-test/data/external_table_p0/iceberg/action/test_iceberg_rewrite_manifests.out
b/regression-test/data/external_table_p0/iceberg/action/test_iceberg_rewrite_manifests.out
index 0d68ae09dc0..e97ac9dec80 100644
---
a/regression-test/data/external_table_p0/iceberg/action/test_iceberg_rewrite_manifests.out
+++
b/regression-test/data/external_table_p0/iceberg/action/test_iceberg_rewrite_manifests.out
@@ -35,6 +35,20 @@
5 item5 500 2024 3
6 item6 600 2024 3
+-- !manifests_by_spec_before --
+0 1
+1 2
+
+-- !spec_id_rewrite --
+2 1
+
+-- !manifests_by_spec_after --
+0 1
+1 1
+
+-- !single_manifest_spec_rewrite --
+0 0
+
-- !after_spec_id_rewrite --
1 item1 100 2024 1 15
2 item2 200 2024 1 16
diff --git
a/regression-test/suites/external_table_p0/iceberg/action/test_iceberg_rewrite_manifests.groovy
b/regression-test/suites/external_table_p0/iceberg/action/test_iceberg_rewrite_manifests.groovy
index 8adb3161066..075b204f232 100644
---
a/regression-test/suites/external_table_p0/iceberg/action/test_iceberg_rewrite_manifests.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/action/test_iceberg_rewrite_manifests.groovy
@@ -238,30 +238,28 @@ suite("test_iceberg_rewrite_manifests", "p0,external") {
sql """INSERT INTO ${db_name}.${spec_id_table} VALUES (4, 'item4', 400,
2024, 3, 18)"""
sql """INSERT INTO ${db_name}.${spec_id_table} VALUES (5, 'item5', 500,
2024, 3, 19)"""
- // Check spec_ids after adding new partition field
- List<List<Object>> allSpecs = sql """
- SELECT partition_spec_id, COUNT(*) as manifest_count
- FROM ${spec_id_table}\$manifests
- GROUP BY partition_spec_id
- ORDER BY partition_spec_id
+ // The first INSERT after the spec change merges the three spec-0
manifests into one, because Iceberg's
+ // merge append compacts the manifests of a spec that the commit does not
write to. So spec 0 has one
+ // manifest and spec 1 has two.
+ order_qt_manifests_by_spec_before """
+ SELECT partition_spec_id, COUNT(*) FROM ${spec_id_table}\$manifests
GROUP BY partition_spec_id
"""
- logger.info("All spec IDs and manifest counts: ${allSpecs}")
-
- if (allSpecs.size() > 0) {
- int targetSpec = allSpecs[0][0] as int
- int targetCount = allSpecs[0][1] as int
- List<List<Object>> specResult = sql """
- ALTER TABLE ${catalog_name}.${db_name}.${spec_id_table}
- EXECUTE rewrite_manifests('spec_id' = '${targetSpec}')
- """
- int specRewritten = specResult[0][0] as int
- assertTrue(specRewritten == targetCount,
- "Should rewrite exactly ${targetCount} manifests for
spec_id=${targetSpec}, got ${specRewritten}")
- qt_after_spec_id_rewrite """SELECT * FROM ${spec_id_table} ORDER BY
id"""
- logger.info("spec_id filtering test completed successfully")
- } else {
- logger.warn("Could not create spec_id, skipping spec_id filtering
test")
- }
+
+ // spec_id limits the rewrite to the two manifests of spec 1, which are
merged into one
+ qt_spec_id_rewrite """
+ ALTER TABLE ${catalog_name}.${db_name}.${spec_id_table}
+ EXECUTE rewrite_manifests('spec_id' = '1')
+ """
+ order_qt_manifests_by_spec_after """
+ SELECT partition_spec_id, COUNT(*) FROM ${spec_id_table}\$manifests
GROUP BY partition_spec_id
+ """
+
+ // A single manifest smaller than the target size is left alone, as
Spark's RewriteManifests action does
+ qt_single_manifest_spec_rewrite """
+ ALTER TABLE ${catalog_name}.${db_name}.${spec_id_table}
+ EXECUTE rewrite_manifests('spec_id' = '0')
+ """
+ qt_after_spec_id_rewrite """SELECT * FROM ${spec_id_table} ORDER BY id"""
//
=====================================================================================
// Test Case 4: rewrite_manifests on empty table (no current snapshot)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
