zachdisc commented on code in PR #9731:
URL: https://github.com/apache/iceberg/pull/9731#discussion_r1501620445
##########
spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java:
##########
@@ -466,6 +474,309 @@ public void testRewriteLargeManifestsPartitionedTable()
throws IOException {
assertThat(newManifests).hasSizeGreaterThanOrEqualTo(2);
}
+ @TestTemplate
+ public void testRewriteManifestsPartitionedTableWithInvalidSortColumns()
throws IOException {
+ PartitionSpec spec =
PartitionSpec.builderFor(SCHEMA).identity("c1").bucket("c3", 10).build();
+ Map<String, String> options = Maps.newHashMap();
+ options.put(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion));
+ options.put(TableProperties.SNAPSHOT_ID_INHERITANCE_ENABLED,
snapshotIdInheritanceEnabled);
+ Table table = TABLES.create(SCHEMA, spec, options, tableLocation);
+
+ SparkActions actions = SparkActions.get();
+
+ // c2 is not a partition column, cannot use for sorting
+ List<String> badSortKeys1 = ImmutableList.of("c1", "c2");
+ assertThatThrownBy(
+ () ->
+ actions
+ .rewriteManifests(table)
+ .rewriteIf(manifest -> true)
+ .sort(badSortKeys1)
+ .execute())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage(
+ "Cannot use custom sort order to rewrite manifests '[c1, c2]'. All
partition columns "
+ + "must be defined in the current partition spec: 0. Choose
from the available "
+ + "partitionable columns: [c3_bucket, c1]");
+
+ // c3_bucket is the correct internal partition name to use, c3 is the
untransformed column name,
+ // sort() expects the hidden partition column names
+ List<String> badSortKeys2 = ImmutableList.of("c1", "c3");
+ assertThatThrownBy(
+ () ->
+ actions
+ .rewriteManifests(table)
+ .rewriteIf(manifest -> true)
+ .sort(badSortKeys2)
+ .execute())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage(
+ "Cannot use custom sort order to rewrite manifests '[c1, c3]'. All
partition columns "
+ + "must be defined in the current partition spec: 0. Choose
from the available "
+ + "partitionable columns: [c3_bucket, c1]");
+ }
+
+ @TestTemplate
+ public void testRewriteManifestsPartitionedTableWithCustomSorting() throws
IOException {
+ Random random = new Random();
+
+ PartitionSpec spec =
+ PartitionSpec.builderFor(SCHEMA).identity("c1").truncate("c2",
3).bucket("c3", 10).build();
+ Table table = TABLES.create(SCHEMA, spec, tableLocation);
+
+ List<DataFile> dataFiles = Lists.newArrayList();
+ for (int fileOrdinal = 0; fileOrdinal < 1000; fileOrdinal++) {
+ dataFiles.add(
+ newDataFile(
+ table,
+ TestHelpers.Row.of(
+ new Object[] {
+ fileOrdinal, String.valueOf(random.nextInt() * 100),
random.nextInt(10)
+ })));
+ }
+ ManifestFile appendManifest = writeManifest(table, dataFiles);
+ table.newFastAppend().appendManifest(appendManifest).commit();
+
+ List<ManifestFile> manifests =
table.currentSnapshot().allManifests(table.io());
+ assertThat(manifests).as("Should have 1 manifests before
rewrite").hasSize(1);
+
+ // Capture the c3 partition's lower and upper bounds - used for later test
assertions
+ Integer c3PartitionMin =
+ Conversions.fromByteBuffer(
+ Types.IntegerType.get(),
manifests.get(0).partitions().get(2).lowerBound());
+ Integer c3PartitionMax =
+ Conversions.fromByteBuffer(
+ Types.IntegerType.get(),
manifests.get(0).partitions().get(2).upperBound());
+
+ // Set the target manifest size to a small value to force splitting
records into multiple files
+ table
+ .updateProperties()
+ .set(
+ TableProperties.MANIFEST_TARGET_SIZE_BYTES,
+ String.valueOf(manifests.get(0).length() / 2))
+ .commit();
+
+ SparkActions actions = SparkActions.get();
+
+ List<String> manifestSortKeys = ImmutableList.of("c3_bucket", "c2_trunc",
"c1");
+ RewriteManifests.Result result =
+ actions
+ .rewriteManifests(table)
+ .rewriteIf(manifest -> true)
+ .sort(manifestSortKeys)
+ .option(RewriteManifestsSparkAction.USE_CACHING, useCaching)
+ .execute();
+
+ table.refresh();
+ List<ManifestFile> newManifests =
table.currentSnapshot().allManifests(table.io());
+
+ assertThat(result.rewrittenManifests()).hasSize(1);
+ assertThat(result.addedManifests()).hasSizeGreaterThanOrEqualTo(2);
+
+ assertThat(newManifests).hasSizeGreaterThanOrEqualTo(2);
+
+ // Rewritten manifests are clustered by c3_bucket - each should contain
only a subset of the
+ // lower and upper bounds
+ // of the partition 'c3'.
+ List<Pair<Integer, Integer>> c3Boundaries =
+ newManifests.stream()
+ .map(manifest -> manifest.partitions().get(2))
+ .sorted(
+ Comparator.comparing(
+ ptn -> Conversions.fromByteBuffer(Types.IntegerType.get(),
ptn.lowerBound())))
+ .map(
+ p ->
+ Pair.of(
+ (Integer)
+
Conversions.fromByteBuffer(Types.IntegerType.get(), p.lowerBound()),
+ (Integer)
+
Conversions.fromByteBuffer(Types.IntegerType.get(), p.upperBound())))
+ .collect(Collectors.toList());
+
+ List<Integer> lowers = c3Boundaries.stream().map(t ->
t.first()).collect(Collectors.toList());
+ List<Integer> uppers = c3Boundaries.stream().map(t ->
t.second()).collect(Collectors.toList());
+
+ // With custom sorting, this looks like
+ // - manifest 1 -> [lower bound = 0, upper bound = 4]
+ // - manifest 2 -> [lower bound = 4, upper bound = 9]
+ // Without the custom sorting, each manifest tracks the full range of c3
upper/lower bounds.
+ // AKA they look like
+ // - manifest 1 -> [lower bound = 0, upper bound = 9]
+ // - manifest 2 -> [lower bound = 0, upper bound = 9]
+ // So the upper bound of the partitions tracked by the first file should
be LEQ the lower bounds
+ // of the second. Etc
+ assertThat(uppers.get(0))
+ .as("Upper bound of first manifest partition should be LEQ lower bound
of second")
+ .isLessThanOrEqualTo(lowers.get(1));
+
+ // Each file should contain less than the full c3 partition span
+ c3Boundaries.forEach(
+ boundary -> {
+ assertThat(boundary.second() - boundary.first())
+ .as("Manifest should contain less than the full range of c3
bucket partitions")
+ .isLessThanOrEqualTo(c3PartitionMax - c3PartitionMin);
+ });
+
+ // c3's Bucket(10) partition means our true lower bound = 0 and true upper
bound is 9. The first
+ // manifest should
+ // include the lower bound of 0, and the last should have the upper bound
of 9
+ assertThat(lowers.get(0))
+ .withFailMessage("Lower bound of first manifest partition should be 0")
+ .isEqualTo(c3PartitionMin);
+ assertThat(uppers.get(uppers.size() - 1))
+ .withFailMessage("Lower bound of first manifest partition should be 0")
+ .isEqualTo(c3PartitionMax);
+ }
+
+ @TestTemplate
+ public void testRewriteManifestsPartitionedTableWithCustomSortFunction()
throws IOException {
+ Random random = new Random();
+
+ PartitionSpec spec =
+ PartitionSpec.builderFor(SCHEMA).identity("c1").truncate("c2",
3).bucket("c3", 10).build();
+ Table table = TABLES.create(SCHEMA, spec, tableLocation);
+
+ List<DataFile> dataFiles = Lists.newArrayList();
+ for (int fileOrdinal = 0; fileOrdinal < 1000; fileOrdinal++) {
+ dataFiles.add(
+ newDataFile(
+ table,
+ TestHelpers.Row.of(
+ new Object[] {
+ fileOrdinal, String.valueOf(random.nextInt() * 100),
random.nextInt(10)
+ })));
+ }
+ ManifestFile appendManifest = writeManifest(table, dataFiles);
+ table.newFastAppend().appendManifest(appendManifest).commit();
+
+ List<ManifestFile> manifests =
table.currentSnapshot().allManifests(table.io());
+ assertThat(manifests).as("Should have 1 manifests before
rewrite").hasSize(1);
+
+ // Capture the c3 partition's lower and upper bounds - used for later test
assertions
+ Integer c3PartitionMin =
+ Conversions.fromByteBuffer(
+ Types.IntegerType.get(),
manifests.get(0).partitions().get(2).lowerBound());
+ Integer c3PartitionMax =
+ Conversions.fromByteBuffer(
+ Types.IntegerType.get(),
manifests.get(0).partitions().get(2).upperBound());
+
+ // Set the target manifest size to a small value to force splitting
records into multiple files
+ table
+ .updateProperties()
+ .set(
+ TableProperties.MANIFEST_TARGET_SIZE_BYTES,
+ String.valueOf(manifests.get(0).length() / 3))
+ .commit();
+
+ SparkActions actions = SparkActions.get();
+
+ // This is the main point of this test!
+ // Let's say I want to sort manifests into buckets 0-4 and 5 - 9
+ // I know my Table and partition Spec, so can do that programmatically
+ Function<DataFile, String> test =
+ (Function<DataFile, String> & Serializable)
Review Comment:
I could not figure out a way to pass a `Function` around for use in UDFs
without declaring it like this at the outset. Looking for ideas here, this part
of the API is a bit clunky.
If you don't declare it this way, we get an exception
```
java.lang.ClassCastException: class
org.apache.iceberg.spark.actions.TestRewriteManifestsAction$$Lambda$1346/0x0000000800cc4040
cannot be cast to class java.io.Serializable
(org.apache.iceberg.spark.actions.TestRewriteManifestsAction$$Lambda$1346/0x0000000800cc4040
is in unnamed module of loader 'app'; java.io.Serializable is in module
java.base of loader 'bootstrap')
```
[I've tried a few variations but it seems like this is by
design](https://stackoverflow.com/questions/25391656/serialization-of-a-lambda-after-its-creation).
So if this requirement is clunky (and I'm not sure how to have our interface
programmatically indicate that this must be a `Serializable` function) then I
think I'd have to require the input to be something we construct and have
people use? Something like
```
interface SerializableFunction<T,R> extends Function<T,R>, Serializable {}
...
private RewriteManifests sort(SerializableFunction<DataFile, String>
function) {
...
}
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
