You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "jbewing (via GitHub)" <gi...@apache.org> on 2026/01/27 04:16:33 UTC
[PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
jbewing opened a new pull request, #15150:
URL: https://github.com/apache/iceberg/pull/15150
This PR updates all writes from Spark 4.0 & 3.5 to set the `sort_order_id` of data files when applicable per the [Iceberg Table Spec](https://iceberg.apache.org/spec/#partition-evolution:~:text=A%20data%20or%20delete%20file%20is%20associated%20with%20a%20sort%20order%20by%20the%20sort%20order%27s%20id%20within%20a%20manifest). I've opened this PR to:
1. Gauge interest in this and validate my approach. If the approach is good, the backport for these changes to Spark 4.0 & 3.5 is clean.
2. If the approach is good, make a decision on whether to backport these changes to Spark 3.4. This patch doesn't apply super cleanly to 3.4 as it stands right now, so I don't want to invest effort into it unless others would find value from it.
3. As a follow up to those PRs, implement the ability to then report this ordering information when applicable to Spark to better inform the query planner of potential optimizations. This is rather useful when used in conjunction with SPJ and I have this running on an Iceberg fork with success. I have done this in a fork to great success. @anuragmantri has since also expressed interest in a parallel patch that they are interested in taking to the finish line after this PR is in. Thank you for the continued interest and support of my changes Anurag! I look forward to collaborating on the follow up PR.
This is a successor to my initial PRs https://github.com/apache/iceberg/pull/13636 & https://github.com/apache/iceberg/pull/14683. The first of which has since been closed for being stale but I later revived for Spark 3.5 & 4.0 in https://github.com/apache/iceberg/pull/14683 at the request of @anuragmantri to complement a follow up change that we both converged on around using this optimization in conjunction w/ some Spark DSv2 APIs to report sort order when possible to optimize downstream query plans. After some discussion w/ @anuragmantr & @RussellSpitzer, I've forward ported the changes to spark 4.1 as the underlying "base" spark version has since changed. I've re-opened this PR as of late there has been some increased interest in this:
- A community member has expressed interest in reviving my older PR https://github.com/apache/iceberg/pull/13636#issuecomment-3505090324
- Other community members have reported this same bug on the issue list https://github.com/apache/iceberg/issues/14531
- Interest by an Iceberg committer @RussellSpitzer
So it appears that there is value to these changes being upstreamed instead of confined to a fork.
### Testing
I've added tests for newer added utility functions and updated existing tests that write data files and compact data files in a sorted manner to verify that we're setting the `sort_order_id` entry in the manifests to the correct value. Additionally, I've used this patch on an internal fork and verified that it correctly sets this field during compaction and normal writes.
Issue: https://github.com/apache/iceberg/issues/13634
cc @anuragmantri & @RussellSpitzer
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2748299868
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkDataWrite.java:
##########
@@ -555,6 +557,55 @@ public void testViewsReturnRecentResults() {
assertThat(actual2).hasSameSizeAs(expected2).isEqualTo(expected2);
}
+ @TestTemplate
+ public void testWriteDataFilesInTableSortOrder() {
+ File parent = temp.resolve(format.toString()).toFile();
+ File location = new File(parent, "test");
+ String targetLocation = locationWithBranch(location);
+
+ HadoopTables tables = new HadoopTables(CONF);
+ PartitionSpec spec = PartitionSpec.unpartitioned();
+ Table table = tables.create(SCHEMA, spec, location.toString());
+
+ table.replaceSortOrder().asc("id").commit();
+
+ List<SimpleRecord> expected = Lists.newArrayListWithCapacity(4000);
+ for (int i = 0; i < 4000; i++) {
+ expected.add(new SimpleRecord(i, "a"));
+ }
+
+ Dataset<Row> df = spark.createDataFrame(expected, SimpleRecord.class);
+
+ df.select("id", "data")
+ .write()
+ .format("iceberg")
+ .option(SparkWriteOptions.WRITE_FORMAT, format.toString())
Review Comment:
I wonder if this is the write place for this test, since we shouldn't have different behavior based on file format ... probably fine
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2760079638
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteOptions.java:
##########
@@ -54,6 +54,7 @@ private SparkWriteOptions() {}
public static final String REWRITTEN_FILE_SCAN_TASK_SET_ID = "rewritten-file-scan-task-set-id";
public static final String OUTPUT_SPEC_ID = "output-spec-id";
+ public static final String OUTPUT_SORT_ORDER_ID = "output-sort-order-id";
Review Comment:
I guess what I was thinking, and let me know if this doesn't make sense. But at some point we decide what the Spark Sort Distribution/ordering is. At that moment we know whether or not the files we write will match a table sort order and we can carry through the id.
Something like
```java
public Integer outputSortOrderId(SparkWriteRequirements writeRequirements) {
if (writeRequirements == null || writeRequirements.ordering().length == 0) {
return SortOrder.unsorted().orderId();
}
org.apache.spark.sql.connector.expressions.SortOrder[] sparkOrdering =
writeRequirements.ordering();
// Find the first table sort order that is satisfied by the Spark ordering
for (SortOrder tableSortOrder : table.sortOrders().values()) {
if (tableSortOrder.isUnsorted()) {
continue;
}
// This is what we would expect this sort order to produce
SortOrder fullSortOrder =
SortOrderUtil.buildSortOrder(table.schema(), table.spec(), tableSortOrder);
// Convert our expected ordering to a spark one
org.apache.spark.sql.connector.expressions.SortOrder[] expectedSparkOrdering =
Spark3Util.toOrdering(fullSortOrder);
if (orderingSatisfies(sparkOrdering, expectedSparkOrdering)) {
return tableSortOrder.orderId();
}
}
// If no match is found, return unsorted
return SortOrder.unsorted().orderId();
}
}
/**
* I asked cursor to generate this, I think there may be much better ways to do this, or at least making a
* dedicated utility method
*/
private boolean orderingSatisfies(
org.apache.spark.sql.connector.expressions.SortOrder[] actualOrdering,
org.apache.spark.sql.connector.expressions.SortOrder[] expectedOrdering) {
if (actualOrdering.length < expectedOrdering.length) {
return false;
}
for (int i = 0; i < expectedOrdering.length; i++) {
org.apache.spark.sql.connector.expressions.SortOrder actual = actualOrdering[i];
org.apache.spark.sql.connector.expressions.SortOrder expected = expectedOrdering[i];
// Check if expressions match
if (!expressionsMatch(actual.expression(), expected.expression())) {
return false;
}
// Check if direction matches
if (actual.direction() != expected.direction()) {
return false;
}
// Check if null order matches
if (actual.nullOrdering() != expected.nullOrdering()) {
return false;
}
}
return true;
}
private boolean expressionsMatch(
org.apache.spark.sql.connector.expressions.Expression expr1,
org.apache.spark.sql.connector.expressions.Expression expr2) {
return expr1.toString().equals(expr2.toString());
}
```
If we do that, then we can just past through the int we want to mark the write with. If Spark decides to do weird things to the required ordering it doesn't matter to us, we are comparing only what they are using to known sort orders
The we just give the id to the writerFactory -- I have more code if you want to see but I just threw it into cursor so it probably still needs some work
```java
private WriterFactory createWriterFactory() {
// broadcast the table metadata as the writer factory will be sent to executors
Broadcast<Table> tableBroadcast =
sparkContext.broadcast(SerializableTableWithSize.copyOf(table));
// Determine the sort order ID that should be assigned to data files
// This is computed in the driver to avoid serialization issues
Integer sortOrderId = writeConf.outputSortOrderId(writeRequirements); // Call our utility
return new WriterFactory(
tableBroadcast,
queryId,
format,
outputSpecId,
targetFileSize,
writeSchema,
dsSchema,
useFanoutWriter,
writeProperties,
sortOrderId); /// Here
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "jbewing (via GitHub)" <gi...@apache.org>.
jbewing commented on PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#issuecomment-4087656589
Addressed all comments in https://github.com/apache/iceberg/pull/15150/changes/45ed1bc88b35547b5b1670393864a7a0b570fc4f & see https://github.com/apache/iceberg/pull/15681 for the split out change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2876106564
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteOptions.java:
##########
@@ -54,6 +54,7 @@ private SparkWriteOptions() {}
public static final String REWRITTEN_FILE_SCAN_TASK_SET_ID = "rewritten-file-scan-task-set-id";
public static final String OUTPUT_SPEC_ID = "output-spec-id";
+ public static final String OUTPUT_SORT_ORDER_ID = "output-sort-order-id";
Review Comment:
Happy to review too, once it is up.
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteOptions.java:
##########
@@ -54,6 +54,7 @@ private SparkWriteOptions() {}
public static final String REWRITTEN_FILE_SCAN_TASK_SET_ID = "rewritten-file-scan-task-set-id";
public static final String OUTPUT_SPEC_ID = "output-spec-id";
+ public static final String OUTPUT_SORT_ORDER_ID = "output-sort-order-id";
Review Comment:
Happy to review too, once implemented.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2966340425
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java:
##########
@@ -171,6 +172,26 @@ public int outputSpecId() {
return outputSpecId;
}
+ public int outputSortOrderId(SparkWriteRequirements writeRequirements) {
+ String explicitId =
+ confParser.stringConf().option(SparkWriteOptions.OUTPUT_SORT_ORDER_ID).parseOptional();
+
+ if (explicitId != null) {
+ int id = Integer.parseInt(explicitId);
+ Preconditions.checkArgument(
+ table.sortOrders().containsKey(id),
+ "Cannot use output sort order id %s because the table does not contain a sort order with that id",
+ id);
+ return id;
+ }
+
+ if (writeRequirements.hasOrdering()) {
+ return table.sortOrder().orderId();
Review Comment:
Thanks for elaborating on the approach. It would be good to be as explicit as possible.
>else, if spark has a sort order set, it must correspond to the iceberg sort order
I'm missing that part. We are making the assumption Iceberg's SortOrder matches Spark's SortOrder and just blindly returning the current table order. Would it make sense to build a Spark SortOrder from the Iceberg SortOrder and verify whether it matches the one from WriteRequirements?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "jbewing (via GitHub)" <gi...@apache.org>.
jbewing commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2985587238
##########
core/src/main/java/org/apache/iceberg/SerializableTable.java:
##########
@@ -80,6 +82,9 @@ protected SerializableTable(Table table) {
Map<Integer, PartitionSpec> specs = table.specs();
specs.forEach((specId, spec) -> specAsJsonMap.put(specId, PartitionSpecParser.toJson(spec)));
this.sortOrderAsJson = SortOrderParser.toJson(table.sortOrder());
+ this.sortOrderAsJsonMap = Maps.newHashMap();
Review Comment:
Not able to do it here (a kryo error get thrown as part of the serialization of this class then), but I believe I addressed your concern just the same with https://github.com/apache/iceberg/pull/15150/commits/cc033550e4ca94559aeddceeb8604a5f8c185db9.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "jbewing (via GitHub)" <gi...@apache.org>.
jbewing commented on PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#issuecomment-3957003794
No worries @RussellSpitzer. I missed some of your comments inadvertently so is absolutely goes both ways :sorry:!
> "We look at what Spark actually ended up sorting by, match it against the table's known sort orders, and pass the matched ID to the file builder."
So this is exactly what I did when I was first designing this PR. I really like the approach. On paper it's way simpler. Your cursor _almost_ one shotted the fundamentals, it missed a few details I outlined in https://github.com/apache/iceberg/pull/15150#discussion_r2850809699. I agree, it's cleaner and it ought to work that way.
That being said and not that it _should_ matter in practice because [per the spec position deletes don't have a defined sort order](https://iceberg.apache.org/spec/#data-file-fields:~:text=Position%20deletes%20are%20required%20to%20be%20sorted%20by%20file%20and%20position%2C%20not%20a%20table%20order%2C%20and%20should%20set%20sort%20order%20id%20to%20null.), is that in the scheme you outlined above, you cannot go backwards from the Spark order to the Iceberg order. Your implementation will return the table sort order for MoR position deletes for a sorted table (the spark ordering is just [POS_DEL]).
A lot of this manifests from the fact that while a Spark ordering may exist when a table sort order isn't set in partitioned write scenarios. Typically, the iceberg sort order fields are a suffix of the Spark ordering, but in many cases the spark ordering isn't == iceberg ordering.
I chose to make the relationship between Spark ordering & Iceberg ordering explicit in my PR to make it impossible for an implicit relationship between the two to inadvertently break things. Yes, it's _slightly_ more verbose, but does reflect the reality that the physical ordering and logical ordering aren't always equal. The Spark ordering != Iceberg ordering was a big learning for me in the codebase—as it was one of my first modifications to iceberg when making this change—and I think the version I've proposed here makes things clearer for readers while being slightly less terse.
All to say: I think both can work. It's a matter of personal preference. Is this blocking for you?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2876102547
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteOptions.java:
##########
@@ -54,6 +54,7 @@ private SparkWriteOptions() {}
public static final String REWRITTEN_FILE_SCAN_TASK_SET_ID = "rewritten-file-scan-task-set-id";
public static final String OUTPUT_SPEC_ID = "output-spec-id";
+ public static final String OUTPUT_SORT_ORDER_ID = "output-sort-order-id";
Review Comment:
I think what @RussellSpitzer is suggesting will work and will likely simplify this PR. I looked through the code that decides Spark ordering based on Iceberg ordering and it will be very hard to capture the ID of the sort order that influenced the final Spark definition.
I would try Russell's idea, this PR is quite involved and it would be great to simplify.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2957408799
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkWriteConf.java:
##########
@@ -606,6 +607,42 @@ public void testDVWriteConf() {
assertThat(writeConf.deleteFileFormat()).isEqualTo(FileFormat.PUFFIN);
}
+ @TestTemplate
+ public void testSortOrderWriteConf() {
+ Table table = validationCatalog.loadTable(tableIdent);
+
+ table.replaceSortOrder().asc("id").commit();
+
+ SparkWriteConf writeConf =
+ new SparkWriteConf(
+ spark,
+ table,
+ new CaseInsensitiveStringMap(
+ ImmutableMap.of(SparkWriteOptions.OUTPUT_SORT_ORDER_ID, "1")));
+
+ assertThat(writeConf.outputSortOrderId(SparkWriteRequirements.EMPTY))
+ .isEqualTo(table.sortOrder().orderId());
+
+ SparkWriteConf writeConfForUnknownSortOrder =
+ new SparkWriteConf(
+ spark,
+ table,
+ new CaseInsensitiveStringMap(
+ ImmutableMap.of(SparkWriteOptions.OUTPUT_SORT_ORDER_ID, "999")));
+
+ assertThatIllegalArgumentException()
+ .isThrownBy(
+ () -> writeConfForUnknownSortOrder.outputSortOrderId(SparkWriteRequirements.EMPTY))
+ .withMessage("Output sort order id 999 is not a valid sort order id for table");
+
+ SparkWriteConf writeConfNoOption = new SparkWriteConf(spark, table);
Review Comment:
Same for here, may as well be another test, I don't think we are saving much in setup duplication
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "anuragmantri (via GitHub)" <gi...@apache.org>.
anuragmantri commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2984952074
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java:
##########
@@ -171,6 +172,26 @@ public int outputSpecId() {
return outputSpecId;
}
+ public int outputSortOrderId(SparkWriteRequirements writeRequirements) {
+ String explicitId =
Review Comment:
Nit: Can we do this in one statement just like outputSpecId() above?
```
Integer explicitId = confParser.intConf()
.option(SparkWriteOptions.OUTPUT_SORT_ORDER_ID)
.parseOptional();
if (explicitId != null) {
// Precondition
```
##########
core/src/main/java/org/apache/iceberg/SerializableTable.java:
##########
@@ -80,6 +82,9 @@ protected SerializableTable(Table table) {
Map<Integer, PartitionSpec> specs = table.specs();
specs.forEach((specId, spec) -> specAsJsonMap.put(specId, PartitionSpecParser.toJson(spec)));
this.sortOrderAsJson = SortOrderParser.toJson(table.sortOrder());
+ this.sortOrderAsJsonMap = Maps.newHashMap();
Review Comment:
Nit: Maybe use immutable map to prevent modification?
```
ImmutableMap.Builder<Integer, String> builder = ImmutableMap.builder();
table.sortOrders().forEach((id, order) -> builder.put(id, SortOrderParser.toJson(order)));
this.sortOrderAsJsonMap = builder.build();
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2760079638
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteOptions.java:
##########
@@ -54,6 +54,7 @@ private SparkWriteOptions() {}
public static final String REWRITTEN_FILE_SCAN_TASK_SET_ID = "rewritten-file-scan-task-set-id";
public static final String OUTPUT_SPEC_ID = "output-spec-id";
+ public static final String OUTPUT_SORT_ORDER_ID = "output-sort-order-id";
Review Comment:
I guess what I was thinking, and let me know if this doesn't make sense. But at some point we decide what the Spark Sort Distribution/ordering is. At that moment we know whether or not the files we write will match a table sort order and we can carry through the id.
Something like
```java
public Integer outputSortOrderId(SparkWriteRequirements writeRequirements) {
if (writeRequirements == null || writeRequirements.ordering().length == 0) {
return SortOrder.unsorted().orderId();
}
org.apache.spark.sql.connector.expressions.SortOrder[] sparkOrdering =
writeRequirements.ordering();
// Find the first table sort order that is satisfied by the Spark ordering
for (SortOrder tableSortOrder : table.sortOrders().values()) {
if (tableSortOrder.isUnsorted()) {
continue;
}
// This is what we would expect this sort order to produce
SortOrder fullSortOrder =
SortOrderUtil.buildSortOrder(table.schema(), table.spec(), tableSortOrder);
// Convert our expected ordering to a spark one
org.apache.spark.sql.connector.expressions.SortOrder[] expectedSparkOrdering =
Spark3Util.toOrdering(fullSortOrder);
if (orderingSatisfies(sparkOrdering, expectedSparkOrdering)) {
return tableSortOrder.orderId();
}
}
// If no match is found, return unsorted
return SortOrder.unsorted().orderId();
}
}
/**
* I asked cursor to generate this, I think there may be much better ways to do this, or at least making a
* dedicated utility method
*/
private boolean orderingSatisfies(
org.apache.spark.sql.connector.expressions.SortOrder[] actualOrdering,
org.apache.spark.sql.connector.expressions.SortOrder[] expectedOrdering) {
if (actualOrdering.length < expectedOrdering.length) {
return false;
}
for (int i = 0; i < expectedOrdering.length; i++) {
org.apache.spark.sql.connector.expressions.SortOrder actual = actualOrdering[i];
org.apache.spark.sql.connector.expressions.SortOrder expected = expectedOrdering[i];
// Check if expressions match
if (!expressionsMatch(actual.expression(), expected.expression())) {
return false;
}
// Check if direction matches
if (actual.direction() != expected.direction()) {
return false;
}
// Check if null order matches
if (actual.nullOrdering() != expected.nullOrdering()) {
return false;
}
}
return true;
}
private boolean expressionsMatch(
org.apache.spark.sql.connector.expressions.Expression expr1,
org.apache.spark.sql.connector.expressions.Expression expr2) {
return expr1.toString().equals(expr2.toString());
}
```
If we do that, then we can just past through the int we want to mark the write with. If Spark decides to do weird things to the required ordering it doesn't matter to us, we are comparing only what they are using to known sort orders
The we just give the id to the writerFactory -- I have more code if you want to see but I just threw it into cursor so it probably still needs some work
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2748305949
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkDataWrite.java:
##########
@@ -555,6 +557,55 @@ public void testViewsReturnRecentResults() {
assertThat(actual2).hasSameSizeAs(expected2).isEqualTo(expected2);
}
+ @TestTemplate
+ public void testWriteDataFilesInTableSortOrder() {
+ File parent = temp.resolve(format.toString()).toFile();
+ File location = new File(parent, "test");
+ String targetLocation = locationWithBranch(location);
+
+ HadoopTables tables = new HadoopTables(CONF);
+ PartitionSpec spec = PartitionSpec.unpartitioned();
+ Table table = tables.create(SCHEMA, spec, location.toString());
+
+ table.replaceSortOrder().asc("id").commit();
+
+ List<SimpleRecord> expected = Lists.newArrayListWithCapacity(4000);
Review Comment:
nit: we don't really need 4000 rows for this test (theoretically we only need 1) but probably wouldn't speed this test up if there where less rows.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2957383944
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteUtil.java:
##########
@@ -85,11 +85,12 @@ public class SparkWriteUtil {
private static final Expression[] PARTITION_FILE_CLUSTERING =
clusterBy(SPEC_ID, PARTITION, FILE_PATH);
- private static final SortOrder[] EMPTY_ORDERING = new SortOrder[0];
Review Comment:
I think we can revert these name changes now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "jbewing (via GitHub)" <gi...@apache.org>.
jbewing commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2978724231
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java:
##########
@@ -171,6 +172,26 @@ public int outputSpecId() {
return outputSpecId;
}
+ public int outputSortOrderId(SparkWriteRequirements writeRequirements) {
+ String explicitId =
+ confParser.stringConf().option(SparkWriteOptions.OUTPUT_SORT_ORDER_ID).parseOptional();
+
+ if (explicitId != null) {
+ int id = Integer.parseInt(explicitId);
+ Preconditions.checkArgument(
+ table.sortOrders().containsKey(id),
+ "Cannot use output sort order id %s because the table does not contain a sort order with that id",
+ id);
+ return id;
+ }
+
+ if (writeRequirements.hasOrdering()) {
+ return table.sortOrder().orderId();
Review Comment:
Does the thread in https://github.com/apache/iceberg/pull/15150#discussion_r2748274402 answer your question @mxm ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "jbewing (via GitHub)" <gi...@apache.org>.
jbewing commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2757050707
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteRequirements.java:
##########
@@ -26,18 +26,32 @@
/** A set of requirements such as distribution and ordering reported to Spark during writes. */
public class SparkWriteRequirements {
+ public static final long NO_ADVISORY_PARTITION_SIZE = 0;
public static final SparkWriteRequirements EMPTY =
- new SparkWriteRequirements(Distributions.unspecified(), new SortOrder[0], 0);
+ new SparkWriteRequirements(
+ Distributions.unspecified(),
+ new SortOrder[0],
+ org.apache.iceberg.SortOrder.unsorted(),
+ NO_ADVISORY_PARTITION_SIZE);
private final Distribution distribution;
private final SortOrder[] ordering;
+ private final org.apache.iceberg.SortOrder icebergOrdering;
private final long advisoryPartitionSize;
SparkWriteRequirements(
Review Comment:
So you probably could get away with just passing the id all the way down and that is actually what is _effectively_ happening here.
We just end up unwrapping from an Id to an Iceberg Sort Order as it's effective at making the code a bit more expressive & readable in some places IMO. SparkWriteRequirement is a nice example of that in that having the Iceberg sort order available makes it _really_ easy to express how the Spark execution sort orders should behave when the Spark ordering doesn't necessarily match the iceberg ordering (and an additional prefix is thrown in there because we're using a range write distribution for example.
I'm happy to unwind this if you don't think that this is the case and the other way is more expressive. I did find in my many iterations of solving this problem "cleanly" that keeping the Sort Orders together—despite the fully qualified class name terribleness—shows the relationship nicely between the two & keeps things concise.
Not passing the Iceberg ordering through is substantially more brittle and prone to breakage (although a bit more concise), however, correctness felt more important the being concise. And passing an Iceberg Sort Order Id down just leads to it being unwrapped from an id in quite a few places and a ton of `if (id == 0 / UNSORTED) {}` checks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "jbewing (via GitHub)" <gi...@apache.org>.
jbewing commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2757073023
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkDataWrite.java:
##########
@@ -555,6 +557,55 @@ public void testViewsReturnRecentResults() {
assertThat(actual2).hasSameSizeAs(expected2).isEqualTo(expected2);
}
+ @TestTemplate
+ public void testWriteDataFilesInTableSortOrder() {
+ File parent = temp.resolve(format.toString()).toFile();
+ File location = new File(parent, "test");
+ String targetLocation = locationWithBranch(location);
+
+ HadoopTables tables = new HadoopTables(CONF);
+ PartitionSpec spec = PartitionSpec.unpartitioned();
+ Table table = tables.create(SCHEMA, spec, location.toString());
+
+ table.replaceSortOrder().asc("id").commit();
+
+ List<SimpleRecord> expected = Lists.newArrayListWithCapacity(4000);
+ for (int i = 0; i < 4000; i++) {
+ expected.add(new SimpleRecord(i, "a"));
+ }
+
+ Dataset<Row> df = spark.createDataFrame(expected, SimpleRecord.class);
+
+ df.select("id", "data")
+ .write()
+ .format("iceberg")
+ .option(SparkWriteOptions.WRITE_FORMAT, format.toString())
+ .mode(SaveMode.Append)
+ .save(location.toString());
+
+ createBranch(table);
Review Comment:
Yeah I copied this test template from the others in this class. I believe they all/most have this and it looks to be leftover from https://github.com/apache/iceberg/pull/6965. Happy to remove, but I was blindly following the "template" that is in place here and going for consistency. So TL;DR: no good reason that we're branching except that almost all the tests in this class do it and I copied that template
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "jbewing (via GitHub)" <gi...@apache.org>.
jbewing commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2957711041
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteRequirements.java:
##########
@@ -26,18 +26,32 @@
/** A set of requirements such as distribution and ordering reported to Spark during writes. */
public class SparkWriteRequirements {
+ public static final long NO_ADVISORY_PARTITION_SIZE = 0;
Review Comment:
Deal! https://github.com/apache/iceberg/pull/15681
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2748298149
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkDataWrite.java:
##########
@@ -555,6 +557,55 @@ public void testViewsReturnRecentResults() {
assertThat(actual2).hasSameSizeAs(expected2).isEqualTo(expected2);
}
+ @TestTemplate
+ public void testWriteDataFilesInTableSortOrder() {
+ File parent = temp.resolve(format.toString()).toFile();
+ File location = new File(parent, "test");
+ String targetLocation = locationWithBranch(location);
+
+ HadoopTables tables = new HadoopTables(CONF);
+ PartitionSpec spec = PartitionSpec.unpartitioned();
+ Table table = tables.create(SCHEMA, spec, location.toString());
+
+ table.replaceSortOrder().asc("id").commit();
+
+ List<SimpleRecord> expected = Lists.newArrayListWithCapacity(4000);
+ for (int i = 0; i < 4000; i++) {
+ expected.add(new SimpleRecord(i, "a"));
+ }
+
+ Dataset<Row> df = spark.createDataFrame(expected, SimpleRecord.class);
+
+ df.select("id", "data")
+ .write()
+ .format("iceberg")
+ .option(SparkWriteOptions.WRITE_FORMAT, format.toString())
+ .mode(SaveMode.Append)
+ .save(location.toString());
+
+ createBranch(table);
+ table.refresh();
+
+ Dataset<Row> result = spark.read().format("iceberg").load(targetLocation);
+
+ List<SimpleRecord> actual =
+ result.orderBy("id").as(Encoders.bean(SimpleRecord.class)).collectAsList();
+ assertThat(actual).hasSameSizeAs(expected).isEqualTo(expected);
+
+ List<DataFile> files = Lists.newArrayList();
Review Comment:
Could use -
table.newScan().planFiles()
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2748285940
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkDataWrite.java:
##########
@@ -555,6 +557,55 @@ public void testViewsReturnRecentResults() {
assertThat(actual2).hasSameSizeAs(expected2).isEqualTo(expected2);
}
+ @TestTemplate
+ public void testWriteDataFilesInTableSortOrder() {
+ File parent = temp.resolve(format.toString()).toFile();
+ File location = new File(parent, "test");
+ String targetLocation = locationWithBranch(location);
+
+ HadoopTables tables = new HadoopTables(CONF);
+ PartitionSpec spec = PartitionSpec.unpartitioned();
+ Table table = tables.create(SCHEMA, spec, location.toString());
Review Comment:
Can't we set the sort order here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "jbewing (via GitHub)" <gi...@apache.org>.
jbewing commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2756994629
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java:
##########
@@ -2657,4 +2705,18 @@ public boolean matches(RewriteFileGroup argument) {
return groupIDs.contains(argument.info().globalIndex());
}
}
+
+ private void dataFilesSortOrderShouldMatchTableSortOrder(Table table) {
+ dataFilesShouldHaveSortOrderIdMatching(table, table.sortOrder());
+ }
+
+ private void dataFilesShouldHaveSortOrderIdMatching(Table table, SortOrder sortOrder) {
+ try (CloseableIterable<FileScanTask> files = table.newScan().planFiles()) {
+ assertThat(files)
+ .extracting(fileScanTask -> fileScanTask.file().sortOrderId())
+ .containsOnly(sortOrder.orderId());
+ } catch (IOException e) {
Review Comment:
👍 will address this in the upcoming revision. I typically just like to get the checked exception over with, but will change it to propagate upwards in the test
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "jbewing (via GitHub)" <gi...@apache.org>.
jbewing commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2757057401
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkDataWrite.java:
##########
@@ -555,6 +557,55 @@ public void testViewsReturnRecentResults() {
assertThat(actual2).hasSameSizeAs(expected2).isEqualTo(expected2);
}
+ @TestTemplate
+ public void testWriteDataFilesInTableSortOrder() {
+ File parent = temp.resolve(format.toString()).toFile();
+ File location = new File(parent, "test");
+ String targetLocation = locationWithBranch(location);
+
+ HadoopTables tables = new HadoopTables(CONF);
+ PartitionSpec spec = PartitionSpec.unpartitioned();
+ Table table = tables.create(SCHEMA, spec, location.toString());
Review Comment:
Can reduce. Full disclosure, this test is a copy-pasta derivative of our friend `testUnpartitionedCreateWithTargetFileSizeViaTableProperties` which is why they share a few similarities 😅
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkDataWrite.java:
##########
@@ -555,6 +557,55 @@ public void testViewsReturnRecentResults() {
assertThat(actual2).hasSameSizeAs(expected2).isEqualTo(expected2);
}
+ @TestTemplate
+ public void testWriteDataFilesInTableSortOrder() {
+ File parent = temp.resolve(format.toString()).toFile();
+ File location = new File(parent, "test");
+ String targetLocation = locationWithBranch(location);
+
+ HadoopTables tables = new HadoopTables(CONF);
+ PartitionSpec spec = PartitionSpec.unpartitioned();
+ Table table = tables.create(SCHEMA, spec, location.toString());
Review Comment:
Can reduce. Full disclosure, this test is a copy-pasta derivative of our friend `testUnpartitionedCreateWithTargetFileSizeViaTableProperties` which is why they share a few similarities 😅
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#issuecomment-4144239637
Merged, Thanks @jbewing for your patience and hard work on this. Thanks @anuragmantri , @aokolnychyi and @mxm for your review.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "jbewing (via GitHub)" <gi...@apache.org>.
jbewing commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2757065826
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkDataWrite.java:
##########
@@ -555,6 +557,55 @@ public void testViewsReturnRecentResults() {
assertThat(actual2).hasSameSizeAs(expected2).isEqualTo(expected2);
}
+ @TestTemplate
+ public void testWriteDataFilesInTableSortOrder() {
+ File parent = temp.resolve(format.toString()).toFile();
+ File location = new File(parent, "test");
+ String targetLocation = locationWithBranch(location);
+
+ HadoopTables tables = new HadoopTables(CONF);
+ PartitionSpec spec = PartitionSpec.unpartitioned();
+ Table table = tables.create(SCHEMA, spec, location.toString());
+
+ table.replaceSortOrder().asc("id").commit();
+
+ List<SimpleRecord> expected = Lists.newArrayListWithCapacity(4000);
+ for (int i = 0; i < 4000; i++) {
+ expected.add(new SimpleRecord(i, "a"));
+ }
+
+ Dataset<Row> df = spark.createDataFrame(expected, SimpleRecord.class);
+
+ df.select("id", "data")
+ .write()
+ .format("iceberg")
+ .option(SparkWriteOptions.WRITE_FORMAT, format.toString())
Review Comment:
Yeah I copied this test template from the others in this class. I believe they all have this. I'm happy to remove if you'd like because as you said it shouldn't matter, but maybe something to say for consistency
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "jbewing (via GitHub)" <gi...@apache.org>.
jbewing commented on PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#issuecomment-3838984849
Addressed your feedback @RussellSpitzer. There are a few open discussion points, let me know your thoughts on how to proceed and I can steer things that way.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "anuragmantri (via GitHub)" <gi...@apache.org>.
anuragmantri commented on PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#issuecomment-3826106547
Thanks @jbewing. @RussellSpitzer - I know the Iceberg summit abstracts are keeping you busy, could you take a look when you get a chance? Thank you!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2982276952
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/SparkShufflingFileRewriteRunner.java:
##########
@@ -113,12 +117,24 @@ public void doRewrite(String groupId, RewriteFileGroup fileGroup) {
spec(fileGroup.outputSpecId()),
fileGroup.expectedOutputFiles()));
+ org.apache.iceberg.SortOrder sortOrderInJobSpec = sortOrder();
+
+ org.apache.iceberg.SortOrder maybeMatchingTableSortOrder =
+ SortOrderUtil.findTableSortOrder(table(), sortOrder());
+
+ if (sortOrderInJobSpec.isSorted() && maybeMatchingTableSortOrder.isUnsorted()) {
+ LOG.warn(
+ "Sort order specified for job {} doesn't match any table sort orders, so going to not mark rewritten files as sorted in the manifest files",
Review Comment:
```suggestion
"Sort order specified for job {} doesn't match any table sort orders, rewritten files will not be marked as sorted in the manifest files",
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "sfc-gh-rspitzer (via GitHub)" <gi...@apache.org>.
sfc-gh-rspitzer commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2957372929
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java:
##########
@@ -171,6 +172,26 @@ public int outputSpecId() {
return outputSpecId;
}
+ public int outputSortOrderId(SparkWriteRequirements writeRequirements) {
+ String explicitId =
+ confParser.stringConf().option(SparkWriteOptions.OUTPUT_SORT_ORDER_ID).parseOptional();
+
+ if (explicitId != null) {
+ int id = Integer.parseInt(explicitId);
+ Preconditions.checkArgument(
+ table.sortOrders().containsKey(id),
+ "Output sort order id %s is not a valid sort order id for table",
Review Comment:
nit: we do "Cannot x because y (optionally fix z)"
So "Cannot use output sort order id %d because the table does not contain a sort order with that id"
or something like that
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#issuecomment-3960365600
Have you checked out RussellSpitzer@c48c6a6? This is much closer to what you are doing just not passing through the actual ordering and also passes all the tests. I basically don't want to have to extend spark requirements, or have special requirements for rewrite files if we can help it.
There are a few more tests there as well which we probably also should be adding to cover checking order assignment for MOR writes and Streaming writes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2957379651
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteRequirements.java:
##########
@@ -26,18 +26,32 @@
/** A set of requirements such as distribution and ordering reported to Spark during writes. */
public class SparkWriteRequirements {
+ public static final long NO_ADVISORY_PARTITION_SIZE = 0;
Review Comment:
I'd just pull this out completely into another PR now. I promise I will review and approve it within 24 hours if you do :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "jbewing (via GitHub)" <gi...@apache.org>.
jbewing commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2978736156
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java:
##########
@@ -171,6 +172,26 @@ public int outputSpecId() {
return outputSpecId;
}
+ public int outputSortOrderId(SparkWriteRequirements writeRequirements) {
+ String explicitId =
+ confParser.stringConf().option(SparkWriteOptions.OUTPUT_SORT_ORDER_ID).parseOptional();
+
+ if (explicitId != null) {
+ int id = Integer.parseInt(explicitId);
+ Preconditions.checkArgument(
+ table.sortOrders().containsKey(id),
+ "Cannot use output sort order id %s because the table does not contain a sort order with that id",
+ id);
+ return id;
+ }
+
+ if (writeRequirements.hasOrdering()) {
+ return table.sortOrder().orderId();
Review Comment:
And if you're interested in following things back in time, you can revisit the original PR that has a fair bit of commentary about this as well: https://github.com/apache/iceberg/pull/14683#discussion_r2560768927
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#issuecomment-4144232204
Talked to Anton who is on vacation (sorry for bothering you Anton) and he said he was good to go.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "jbewing (via GitHub)" <gi...@apache.org>.
jbewing commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2757061213
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkDataWrite.java:
##########
@@ -555,6 +557,55 @@ public void testViewsReturnRecentResults() {
assertThat(actual2).hasSameSizeAs(expected2).isEqualTo(expected2);
}
+ @TestTemplate
+ public void testWriteDataFilesInTableSortOrder() {
+ File parent = temp.resolve(format.toString()).toFile();
+ File location = new File(parent, "test");
+ String targetLocation = locationWithBranch(location);
+
+ HadoopTables tables = new HadoopTables(CONF);
+ PartitionSpec spec = PartitionSpec.unpartitioned();
+ Table table = tables.create(SCHEMA, spec, location.toString());
+
+ table.replaceSortOrder().asc("id").commit();
+
+ List<SimpleRecord> expected = Lists.newArrayListWithCapacity(4000);
Review Comment:
Can reduce. Full disclosure, this test is a copy-pasta derivative of our friend `testUnpartitionedCreateWithTargetFileSizeViaTableProperties` which is why they share a few similarities 😅
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2748254300
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteRequirements.java:
##########
@@ -26,18 +26,32 @@
/** A set of requirements such as distribution and ordering reported to Spark during writes. */
public class SparkWriteRequirements {
+ public static final long NO_ADVISORY_PARTITION_SIZE = 0;
Review Comment:
How is this change related to the PR?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2754897432
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java:
##########
@@ -2657,4 +2705,18 @@ public boolean matches(RewriteFileGroup argument) {
return groupIDs.contains(argument.info().globalIndex());
}
}
+
+ private void dataFilesSortOrderShouldMatchTableSortOrder(Table table) {
+ dataFilesShouldHaveSortOrderIdMatching(table, table.sortOrder());
+ }
+
+ private void dataFilesShouldHaveSortOrderIdMatching(Table table, SortOrder sortOrder) {
+ try (CloseableIterable<FileScanTask> files = table.newScan().planFiles()) {
+ assertThat(files)
+ .extracting(fileScanTask -> fileScanTask.file().sortOrderId())
+ .containsOnly(sortOrder.orderId());
+ } catch (IOException e) {
Review Comment:
it's ok to just rethrow in the tests, we do that pattern pretty frequently
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "jbewing (via GitHub)" <gi...@apache.org>.
jbewing commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2757027071
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteOptions.java:
##########
@@ -54,6 +54,7 @@ private SparkWriteOptions() {}
public static final String REWRITTEN_FILE_SCAN_TASK_SET_ID = "rewritten-file-scan-task-set-id";
public static final String OUTPUT_SPEC_ID = "output-spec-id";
+ public static final String OUTPUT_SORT_ORDER_ID = "output-sort-order-id";
Review Comment:
Absolutely, but unfortunately there are a few core write paths that don't necessarily "mirror" the table sort order all the time so sometimes it needs to be customized. Those are:
1. Rewrite Data Files (changes can be seen above in `SparkShufflingFileRewriteRunner`, but TL;DR: the default is to use the table sort order, but an operator can pass in any sort order they'd like to this job and it _will_ comply
2. My memory is a bit fuzzier at this point, but I believe there are a few write time options you can pass to essentially ignore table ordering. The ones that come to mind here that I can remember are: "fanout-enabled & write-distribution=none (in these cases the Spark sort order will match the Iceberg sort order though). SparkWriteUtil handles a lot of the heavy lifting here as far as how these interact with Spark <-> Iceberg sort orders, but in a lot modes it's actually not possible to directly extract the iceberg sort order intent from the Spark execution plan as Spark will sometimes prepend partition keys to the sort order keys in certain operations.
Worth taking a look around SparkWriteUtil as I said, but the exciting thing I learned on this PR adventure is: Iceberg Table sort order is not necessarily equal to Iceberg operation sort order which is also not necessarily equal to Spark sort order
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#issuecomment-3954311799
> I think @jbewing did it this way because Iceberg may request Spark to order by fields outside of just the table sort-order-id (for example, sort by row positions, etc.). We need the transform to convert them to Iceberg sort order id metadata.
>
This is why I suggest the other approach since it's more genericy. Instead of assuming what gets passed through is what is set, we can always just look at what Spark has chosen to use as it's distribution and ordering and figure out if that maps to a given sort Id.
So the approach i'm describing is essentially
"We look at what Spark actually ended up sorting by, match it against the table's known sort orders, and pass the matched ID to the file builder."
While the current approach is
"We decide the sort order ID up front when planning the write, then carry it alongside Spark's ordering all the way to the file builder."
I did a full example here with some more tests - https://github.com/RussellSpitzer/iceberg/commit/c48c6a600379a773b15fc68b933a0ce51dfe994b if you want to check it out
I think we may have a larger disagreement here, and sorry for the amount of time it's taken me in between checks. Let me see if we can get @szehon-ho or @aokolnychyi to also take a look here. They are both very focused on the spark side and probably have some opinions on the right way to pass this through.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "jbewing (via GitHub)" <gi...@apache.org>.
jbewing commented on PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#issuecomment-4035625076
Sorry for the delay here @RussellSpitzer . I've added some additional test coverage at your suggestion. I just solved some merge conflicts with master. And I've now gotten a chance to read your approach.
The way I see it, your changeset has followed an extremely similar arc to mine. Started with trying to derive iceberg order from spark order. Then, moved to using the iceberg table ordering at write time (in combo w/ the various other misc options out there like explicitly setting a sort order for compactions/rewrites).
They diverge by about ~60 LoC with the primary difference between them being: my changeset is a bit more explicit (e.g. uses stronger types & less implicit coupling between iceberg sort order & spark sort). And at this point, that's the total sum of the differences.
You're a maintainer so you absolutely get the final call here, but as far as I see it they both have their advantages.
I think my approach is probably slightly friendlier to non-long-term readers of the codebase, but it does also come w/ the downside of being slightly less performant in terms of serializing a slightly larger object to the executors. In practice, this has been running in production for us on larger tables (hundreds of TB) that get partially rewritten many many times per day and I've never noticed the increased overhead on a flamegraph, so I don't think that's a crazy concern.
But yeah otherwise, I'm happy to change to yours if you think the 60 LoC difference / minute difference in approach makes a huge difference in readability. Having read both: we both converged on the same place for this changeset. They are functionally identical and just two different styles of expressing the same concept.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "sfc-gh-rspitzer (via GitHub)" <gi...@apache.org>.
sfc-gh-rspitzer commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2957359081
##########
core/src/main/java/org/apache/iceberg/util/SortOrderUtil.java:
##########
@@ -46,6 +46,23 @@ public static SortOrder buildSortOrder(Table table, SortOrder sortOrder) {
return buildSortOrder(table.schema(), table.spec(), sortOrder);
}
+ /**
+ * Attempts to match a user-supplied {@link SortOrder} with an equivalent sort order from a {@link
+ * Table}.
+ *
+ * @param table the table to try and match the sort order against
+ * @param userSuppliedSortOrder the user supplied sort order to try and match with a table sort
+ * order
+ * @return the matching {@link SortOrder} from the table (with the orderId set) or {@link
+ * SortOrder#unsorted()} if no match is found.
+ */
+ public static SortOrder maybeFindTableSortOrder(Table table, SortOrder userSuppliedSortOrder) {
Review Comment:
nit: findTableSortOrder? We tend not to use maybes in lookups methods.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2957432916
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java:
##########
@@ -1695,10 +1698,54 @@ public void testSortCustomSortOrderRequiresRepartition() {
shouldHaveMultipleFiles(table);
shouldHaveLastCommitUnsorted(table, "c2");
shouldHaveLastCommitSorted(table, "c3");
+ dataFilesShouldHaveSortOrderIdMatching(table, SortOrder.unsorted());
}
@TestTemplate
- public void testAutoSortShuffleOutput() {
+ public void testSortPastTableSortOrderGetsAppliedToFiles() throws IOException {
+ int partitions = 4;
+ Table table = createTable();
+ writeRecords(20, SCALE, partitions);
+ shouldHaveLastCommitUnsorted(table, "c3");
+
+ table.updateSpec().addField("c1").commit();
+
+ table.replaceSortOrder().asc("c3").commit();
+ SortOrder c3SortOrder = table.sortOrder();
+
+ table.replaceSortOrder().asc("c2").commit();
+ shouldHaveFiles(table, 20);
+
+ List<Object[]> originalData = currentData();
+ long dataSizeBefore = testDataSize(table);
+
+ RewriteDataFiles.Result result =
+ basicRewrite(table)
+ .sort(SortOrder.builderFor(table.schema()).asc("c3").build())
+ .option(SizeBasedFileRewritePlanner.REWRITE_ALL, "true")
+ .option(
+ RewriteDataFiles.TARGET_FILE_SIZE_BYTES,
Review Comment:
as a minor speedup to this test you don't need to mess with target file size. We do this in the other tests because we want to make sure there are multiple files, but in this test we only care about the sort order being set. So theoretically you could just make it a single file compaction and just check that it's properly labeled.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#issuecomment-4087252215
Structurally I think this is all there, I think we have a few remaining test nits and some reversions of renames from previous prs and one extraction to a different PR but pretty close. @aokolnychyi Do you want to take a pass as well?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2754918198
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteRequirements.java:
##########
@@ -26,18 +26,32 @@
/** A set of requirements such as distribution and ordering reported to Spark during writes. */
public class SparkWriteRequirements {
+ public static final long NO_ADVISORY_PARTITION_SIZE = 0;
public static final SparkWriteRequirements EMPTY =
- new SparkWriteRequirements(Distributions.unspecified(), new SortOrder[0], 0);
+ new SparkWriteRequirements(
+ Distributions.unspecified(),
+ new SortOrder[0],
+ org.apache.iceberg.SortOrder.unsorted(),
+ NO_ADVISORY_PARTITION_SIZE);
private final Distribution distribution;
private final SortOrder[] ordering;
+ private final org.apache.iceberg.SortOrder icebergOrdering;
private final long advisoryPartitionSize;
SparkWriteRequirements(
Review Comment:
I'm not sure we should be passing the Iceberg ordering all the way down, I feel like all the writer actually needs to know is the ID it should be using in the Datafile for the sort order used and we should be doing all the work to determine what sortorder is and how to pass it as a spark ordering in code calling this
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "jbewing (via GitHub)" <gi...@apache.org>.
jbewing commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2756986601
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteRequirements.java:
##########
@@ -26,18 +26,32 @@
/** A set of requirements such as distribution and ordering reported to Spark during writes. */
public class SparkWriteRequirements {
+ public static final long NO_ADVISORY_PARTITION_SIZE = 0;
Review Comment:
Happy to revert if you'd like, but I believe that the:
1. Extraction into a named constant
2. Shifting from read-time calculated many times to write-once calculated
is a net good change that I made along the way with adding the additional instance field for Iceberg sort order.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "jbewing (via GitHub)" <gi...@apache.org>.
jbewing commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2756984213
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteRequirements.java:
##########
@@ -26,18 +26,32 @@
/** A set of requirements such as distribution and ordering reported to Spark during writes. */
public class SparkWriteRequirements {
+ public static final long NO_ADVISORY_PARTITION_SIZE = 0;
Review Comment:
This isn't directly related, but given that I was refactoring SparkWriteRequirements I stumbled upon this unnamed constant e.g. 0 and decided to pay it forwards into a named constant since I was modifying this class anyway and it took me a minute to figure out what 0 meant without context
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2754900136
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java:
##########
@@ -2657,4 +2705,18 @@ public boolean matches(RewriteFileGroup argument) {
return groupIDs.contains(argument.info().globalIndex());
}
}
+
+ private void dataFilesSortOrderShouldMatchTableSortOrder(Table table) {
+ dataFilesShouldHaveSortOrderIdMatching(table, table.sortOrder());
+ }
+
+ private void dataFilesShouldHaveSortOrderIdMatching(Table table, SortOrder sortOrder) {
+ try (CloseableIterable<FileScanTask> files = table.newScan().planFiles()) {
+ assertThat(files)
+ .extracting(fileScanTask -> fileScanTask.file().sortOrderId())
+ .containsOnly(sortOrder.orderId());
+ } catch (IOException e) {
Review Comment:
```java
private void dataFilesShould...() Throws IOException {
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2748280703
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkDataWrite.java:
##########
@@ -555,6 +557,55 @@ public void testViewsReturnRecentResults() {
assertThat(actual2).hasSameSizeAs(expected2).isEqualTo(expected2);
}
+ @TestTemplate
+ public void testWriteDataFilesInTableSortOrder() {
+ File parent = temp.resolve(format.toString()).toFile();
+ File location = new File(parent, "test");
+ String targetLocation = locationWithBranch(location);
+
+ HadoopTables tables = new HadoopTables(CONF);
+ PartitionSpec spec = PartitionSpec.unpartitioned();
+ Table table = tables.create(SCHEMA, spec, location.toString());
+
+ table.replaceSortOrder().asc("id").commit();
+
+ List<SimpleRecord> expected = Lists.newArrayListWithCapacity(4000);
+ for (int i = 0; i < 4000; i++) {
+ expected.add(new SimpleRecord(i, "a"));
+ }
+
+ Dataset<Row> df = spark.createDataFrame(expected, SimpleRecord.class);
+
+ df.select("id", "data")
+ .write()
+ .format("iceberg")
+ .option(SparkWriteOptions.WRITE_FORMAT, format.toString())
+ .mode(SaveMode.Append)
+ .save(location.toString());
+
+ createBranch(table);
Review Comment:
Why are we branching?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "anuragmantri (via GitHub)" <gi...@apache.org>.
anuragmantri commented on PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#issuecomment-3837498559
> I could definitely be convinced otherwise, but it seems odd to me that we are carrying around two different sort objects in the writer.
I think @jbewing did it this way because Iceberg may request Spark to order by fields outside of just the table sort-order-id (for example, sort by row positions, etc.). We need the transform to convert them to Iceberg sort order id metadata.
But I agree that carrying both the Spark and Iceberg sort orders is odd. Would it be too restrictive to put a constraint that we will only sort by table sort order id? Would it be simpler to always materialize the table sort order id in the metadata?
(We don't do this validation currently. For example, we can pass any arbitrary field to RewriteDataFiles, and with this PR, it will be persisted in the metadata.)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2980742218
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java:
##########
@@ -171,6 +172,26 @@ public int outputSpecId() {
return outputSpecId;
}
+ public int outputSortOrderId(SparkWriteRequirements writeRequirements) {
+ String explicitId =
+ confParser.stringConf().option(SparkWriteOptions.OUTPUT_SORT_ORDER_ID).parseOptional();
+
+ if (explicitId != null) {
+ int id = Integer.parseInt(explicitId);
+ Preconditions.checkArgument(
+ table.sortOrders().containsKey(id),
+ "Cannot use output sort order id %s because the table does not contain a sort order with that id",
+ id);
+ return id;
+ }
+
+ if (writeRequirements.hasOrdering()) {
+ return table.sortOrder().orderId();
Review Comment:
Thanks! I'll do some more digging. It looks like Spark performs its own sort optimizations which may be different from the original table sort order.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "jbewing (via GitHub)" <gi...@apache.org>.
jbewing commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2978732314
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java:
##########
@@ -171,6 +172,26 @@ public int outputSpecId() {
return outputSpecId;
}
+ public int outputSortOrderId(SparkWriteRequirements writeRequirements) {
+ String explicitId =
+ confParser.stringConf().option(SparkWriteOptions.OUTPUT_SORT_ORDER_ID).parseOptional();
+
+ if (explicitId != null) {
+ int id = Integer.parseInt(explicitId);
+ Preconditions.checkArgument(
+ table.sortOrders().containsKey(id),
+ "Cannot use output sort order id %s because the table does not contain a sort order with that id",
+ id);
+ return id;
+ }
+
+ if (writeRequirements.hasOrdering()) {
+ return table.sortOrder().orderId();
Review Comment:
But essentially they often don't match in practice which is fine. There are various write configs + table configs that can change this. Additionally, using something like partitioning w/ ordering produces spark sort orders that have the partition keys & the prefix of the spark sort order followed by the actual iceberg sort order as the suffix
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#issuecomment-3985937898
I think those edge cases apply to the inline code I posted in the comment (the orderingSatisfies approach), not the actual commit at RussellSpitzer@c48c6a6. I abandoned that approach and Instead tried to boil it down to one method.
outputSortOrderId(writeRequirements) just checks: did we ask Spark to sort? If hasOrdering() is true we know it came from the table sort order (that's what SparkWriteUtil builds), so we use table.sortOrder().orderId(). No suffix matching, no iterating table sort orders.
For rewrites, the OUTPUT_SORT_ORDER_ID path is the same as yours. This avoids the class extension for SparkRequirements and removes the new requirements for the Sort Compaction case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "sfc-gh-rspitzer (via GitHub)" <gi...@apache.org>.
sfc-gh-rspitzer commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2957349013
##########
core/src/main/java/org/apache/iceberg/SerializableTable.java:
##########
@@ -240,7 +245,21 @@ public SortOrder sortOrder() {
@Override
public Map<Integer, SortOrder> sortOrders() {
- return lazyTable().sortOrders();
+ if (lazySortOrders == null) {
Review Comment:
This makes sense to me as a good fix for now. We already do this for a bunch of other things like (specs) and it should be relatively lightweight.
I'd ideally also still like to just get the ability to pass the ID directly into the factory but that could be a follow up issue someone can work on.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2957445843
##########
core/src/test/java/org/apache/iceberg/util/TestSortOrderUtil.java:
##########
@@ -287,4 +287,68 @@ public void testSortOrderClusteringWithRedundantPartitionFieldsMissing() {
.as("Should add spec fields as prefix")
.isEqualTo(expected);
}
+
+ @Test
+ public void testFindSortOrderForTable() {
+ PartitionSpec spec = PartitionSpec.unpartitioned();
+ SortOrder order = SortOrder.builderFor(SCHEMA).withOrderId(1).asc("id", NULLS_LAST).build();
+ TestTables.TestTable table = TestTables.create(tableDir, "test", SCHEMA, spec, order, 2);
+
+ SortOrder tableSortOrder = table.sortOrder();
+
+ SortOrder actualOrder = SortOrderUtil.maybeFindTableSortOrder(table, tableSortOrder);
+
+ assertThat(actualOrder).as("Should find current table sort order").isEqualTo(table.sortOrder());
+ }
+
+ @Test
+ public void testFindSortOrderForTableWithoutFieldId() {
+ PartitionSpec spec = PartitionSpec.unpartitioned();
+ SortOrder order = SortOrder.builderFor(SCHEMA).withOrderId(1).asc("id", NULLS_LAST).build();
+ TestTables.TestTable table = TestTables.create(tableDir, "test", SCHEMA, spec, order, 2);
+
+ SortOrder userSuppliedOrder =
+ SortOrder.builderFor(table.schema()).asc("id", NULLS_LAST).build();
+
+ SortOrder actualOrder = SortOrderUtil.maybeFindTableSortOrder(table, userSuppliedOrder);
+
+ assertThat(actualOrder).as("Should find current table sort order").isEqualTo(table.sortOrder());
+ }
+
+ @Test
+ public void testFindSortOrderForTableThatIsNotCurrentOrder() {
+ PartitionSpec spec = PartitionSpec.unpartitioned();
+ SortOrder order = SortOrder.builderFor(SCHEMA).withOrderId(1).asc("id", NULLS_LAST).build();
+ TestTables.TestTable table = TestTables.create(tableDir, "test", SCHEMA, spec, order, 2);
+
+ table.replaceSortOrder().asc("data").desc("ts").commit();
+
+ SortOrder userSuppliedOrder =
+ SortOrder.builderFor(table.schema()).asc("id", NULLS_LAST).build();
+
+ SortOrder actualOrder = SortOrderUtil.maybeFindTableSortOrder(table, userSuppliedOrder);
+
+ assertThat(actualOrder)
+ .as("Should find first sorted table sort order")
+ .isEqualTo(table.sortOrders().get(1));
+ }
+
+ @Test
+ public void testReturnsEmptyForFindingNonMatchingSortOrder() {
Review Comment:
nit: testReturnsUnsortedForMissingSortOrder
Since it isn't empty, just unsorted.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2961175351
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkDataWrite.java:
##########
@@ -555,6 +557,55 @@ public void testViewsReturnRecentResults() {
assertThat(actual2).hasSameSizeAs(expected2).isEqualTo(expected2);
}
+ @TestTemplate
+ public void testWriteDataFilesInTableSortOrder() {
+ File parent = temp.resolve(format.toString()).toFile();
+ File location = new File(parent, "test");
+ String targetLocation = locationWithBranch(location);
+
+ HadoopTables tables = new HadoopTables(CONF);
+ PartitionSpec spec = PartitionSpec.unpartitioned();
+ Table table = tables.create(SCHEMA, spec, location.toString());
+
+ table.replaceSortOrder().asc("id").commit();
+
+ List<SimpleRecord> expected = Lists.newArrayListWithCapacity(4000);
+ for (int i = 0; i < 4000; i++) {
+ expected.add(new SimpleRecord(i, "a"));
+ }
+
+ Dataset<Row> df = spark.createDataFrame(expected, SimpleRecord.class);
+
+ df.select("id", "data")
+ .write()
+ .format("iceberg")
+ .option(SparkWriteOptions.WRITE_FORMAT, format.toString())
+ .mode(SaveMode.Append)
+ .save(location.toString());
+
+ createBranch(table);
Review Comment:
+1 We should not be branching here (and elsewhere) for no reason.
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java:
##########
@@ -171,6 +172,26 @@ public int outputSpecId() {
return outputSpecId;
}
+ public int outputSortOrderId(SparkWriteRequirements writeRequirements) {
+ String explicitId =
+ confParser.stringConf().option(SparkWriteOptions.OUTPUT_SORT_ORDER_ID).parseOptional();
+
+ if (explicitId != null) {
+ int id = Integer.parseInt(explicitId);
+ Preconditions.checkArgument(
+ table.sortOrders().containsKey(id),
+ "Cannot use output sort order id %s because the table does not contain a sort order with that id",
+ id);
+ return id;
+ }
+
+ if (writeRequirements.hasOrdering()) {
+ return table.sortOrder().orderId();
Review Comment:
Why are we using the table's sort order and not the ordering from the `writeRequirements`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer merged PR #15150:
URL: https://github.com/apache/iceberg/pull/15150
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "jbewing (via GitHub)" <gi...@apache.org>.
jbewing commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2850809699
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteOptions.java:
##########
@@ -54,6 +54,7 @@ private SparkWriteOptions() {}
public static final String REWRITTEN_FILE_SCAN_TASK_SET_ID = "rewritten-file-scan-task-set-id";
public static final String OUTPUT_SPEC_ID = "output-spec-id";
+ public static final String OUTPUT_SORT_ORDER_ID = "output-sort-order-id";
Review Comment:
Missed this one a few weeks ago, sorry about that. But I did totally start here. I can provide a bug fix that makes your cursor impl pass a few more test cases which is basically that you want to walk the spark actual sort order and your expected iceberg sort order from the end of the actual spark sort order rather than starting at the front. The different is subtle but it accommodates cases like this: https://github.com/apache/iceberg/blob/9534c9b3adc29d127ecc541ce131f49fd72f1980/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteUtil.java#L210-L216
You also want to be quite careful about the order in which you iterate the table sort orders. I found that an explicit iteration order of:
1. Prefer the active sort order first (this may or may not be the sort order with the highest id)
2. Then prefer going in descending id order of the inactive sort orders
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2748274402
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteOptions.java:
##########
@@ -54,6 +54,7 @@ private SparkWriteOptions() {}
public static final String REWRITTEN_FILE_SCAN_TASK_SET_ID = "rewritten-file-scan-task-set-id";
public static final String OUTPUT_SPEC_ID = "output-spec-id";
+ public static final String OUTPUT_SORT_ORDER_ID = "output-sort-order-id";
Review Comment:
This is kind of broaching other concerns. I think it would make sense to start with just supporting writing the current table output-sort-order, then work on arbitrary sort order in a followup. Unless I'm reading this incorrectly?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2759856755
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkDataWrite.java:
##########
@@ -555,6 +557,55 @@ public void testViewsReturnRecentResults() {
assertThat(actual2).hasSameSizeAs(expected2).isEqualTo(expected2);
}
+ @TestTemplate
+ public void testWriteDataFilesInTableSortOrder() {
+ File parent = temp.resolve(format.toString()).toFile();
+ File location = new File(parent, "test");
+ String targetLocation = locationWithBranch(location);
+
+ HadoopTables tables = new HadoopTables(CONF);
+ PartitionSpec spec = PartitionSpec.unpartitioned();
+ Table table = tables.create(SCHEMA, spec, location.toString());
+
+ table.replaceSortOrder().asc("id").commit();
+
+ List<SimpleRecord> expected = Lists.newArrayListWithCapacity(4000);
+ for (int i = 0; i < 4000; i++) {
+ expected.add(new SimpleRecord(i, "a"));
+ }
+
+ Dataset<Row> df = spark.createDataFrame(expected, SimpleRecord.class);
+
+ df.select("id", "data")
+ .write()
+ .format("iceberg")
+ .option(SparkWriteOptions.WRITE_FORMAT, format.toString())
+ .mode(SaveMode.Append)
+ .save(location.toString());
+
+ createBranch(table);
Review Comment:
Yeah this whole file is full of some things i'm a little confused about, I don't think it's worth copying them. We can put a followup issue to clean up this file.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "jbewing (via GitHub)" <gi...@apache.org>.
jbewing commented on PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#issuecomment-4159603823
Thanks for all the help @RussellSpitzer, @anuragmantri , @aokolnychyi , @mxm ! See https://github.com/apache/iceberg/pull/15832 for the backport of these changes to Spark 4.0 & Spark 3.5
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#issuecomment-4050029493
I think we've tried to talk about this a bit before, but the main concern we have is changing existing classes and adding additional ones in order to essentially carry an integer through the pipeline.
We are always attempting to make the least invasive changes possible and I think in this case we are modifying a class with a clear purpose (carry spark specific requirements which change how Spark performs the write) with an iceberg concern (how do we annotate the files produced).
This is not a performance issue but an architectural one in the codebase. That's the kind of thing @aokolnychyi and I are mentioning. This is a pretty opinionated project on that sort of thing. I know this feels a little arbitrary but it is something we take pretty seriously.
Ideally, we keep these concerns separate unless there is a really strong argument as to why they belong together.
I'm glad to keep working with you on this PR but we need to pass the Iceberg information in its own class/pathway/container or just as an int. Anytime we have a "This and That" class it's going to be an issue.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2957404659
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkWriteConf.java:
##########
@@ -606,6 +607,42 @@ public void testDVWriteConf() {
assertThat(writeConf.deleteFileFormat()).isEqualTo(FileFormat.PUFFIN);
}
+ @TestTemplate
+ public void testSortOrderWriteConf() {
+ Table table = validationCatalog.loadTable(tableIdent);
+
+ table.replaceSortOrder().asc("id").commit();
+
+ SparkWriteConf writeConf =
+ new SparkWriteConf(
+ spark,
+ table,
+ new CaseInsensitiveStringMap(
+ ImmutableMap.of(SparkWriteOptions.OUTPUT_SORT_ORDER_ID, "1")));
+
+ assertThat(writeConf.outputSortOrderId(SparkWriteRequirements.EMPTY))
+ .isEqualTo(table.sortOrder().orderId());
+
+ SparkWriteConf writeConfForUnknownSortOrder =
Review Comment:
nit: could be another test since this is assertion on a different code path. There isn't a lot of setup for this one since you just need to load a table so I would probably keep it separated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#issuecomment-4119035621
@mxm + @aokolnychyi Any blockers on this? I think we are good to go.
We can clean up the testSpark classes in a followup , I did a quick check and I think we can remove like 100~ lines from that test class but I'd rather we do it all at once in a fresh pr.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#issuecomment-4119127400
Sounds good. No blockers from my side.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2876102547
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteOptions.java:
##########
@@ -54,6 +54,7 @@ private SparkWriteOptions() {}
public static final String REWRITTEN_FILE_SCAN_TASK_SET_ID = "rewritten-file-scan-task-set-id";
public static final String OUTPUT_SPEC_ID = "output-spec-id";
+ public static final String OUTPUT_SORT_ORDER_ID = "output-sort-order-id";
Review Comment:
I think what @RussellSpitzer is suggesting will work and will likely simplify this PR. I looked through the code that decides Spark ordering based on Iceberg ordering and it will be very hard to capture the ID of the sort order that influenced the final Spark definition. Given that is not an option, then comparing the Spark order to table orders somewhere after the requirements are built is the only option.
I would try Russell's idea, this PR is quite involved and it would be great to simplify.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2876102547
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteOptions.java:
##########
@@ -54,6 +54,7 @@ private SparkWriteOptions() {}
public static final String REWRITTEN_FILE_SCAN_TASK_SET_ID = "rewritten-file-scan-task-set-id";
public static final String OUTPUT_SPEC_ID = "output-spec-id";
+ public static final String OUTPUT_SORT_ORDER_ID = "output-sort-order-id";
Review Comment:
I think what @RussellSpitzer is suggesting will work and will likely simplify this PR. I looked through the code that decides Spark ordering based on Iceberg ordering and it will be very hard to capture the ID of the sort order that influenced the final Spark definition. Given that is not an option, then comparing the Spark order to table orders is the only option.
I would try Russell's idea, this PR is quite involved and it would be great to simplify.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "jbewing (via GitHub)" <gi...@apache.org>.
jbewing commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2963796237
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkDataWrite.java:
##########
@@ -555,6 +557,55 @@ public void testViewsReturnRecentResults() {
assertThat(actual2).hasSameSizeAs(expected2).isEqualTo(expected2);
}
+ @TestTemplate
+ public void testWriteDataFilesInTableSortOrder() {
+ File parent = temp.resolve(format.toString()).toFile();
+ File location = new File(parent, "test");
+ String targetLocation = locationWithBranch(location);
+
+ HadoopTables tables = new HadoopTables(CONF);
+ PartitionSpec spec = PartitionSpec.unpartitioned();
+ Table table = tables.create(SCHEMA, spec, location.toString());
+
+ table.replaceSortOrder().asc("id").commit();
+
+ List<SimpleRecord> expected = Lists.newArrayListWithCapacity(4000);
+ for (int i = 0; i < 4000; i++) {
+ expected.add(new SimpleRecord(i, "a"));
+ }
+
+ Dataset<Row> df = spark.createDataFrame(expected, SimpleRecord.class);
+
+ df.select("id", "data")
+ .write()
+ .format("iceberg")
+ .option(SparkWriteOptions.WRITE_FORMAT, format.toString())
+ .mode(SaveMode.Append)
+ .save(location.toString());
+
+ createBranch(table);
Review Comment:
See https://github.com/apache/iceberg/pull/15150/commits/e04a4987b0ea4dec7714274897c7a60f0174360d
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Set data file sort_order_id in manifest for writes from Spark [iceberg]
Posted by "jbewing (via GitHub)" <gi...@apache.org>.
jbewing commented on code in PR #15150:
URL: https://github.com/apache/iceberg/pull/15150#discussion_r2963807856
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java:
##########
@@ -171,6 +172,26 @@ public int outputSpecId() {
return outputSpecId;
}
+ public int outputSortOrderId(SparkWriteRequirements writeRequirements) {
+ String explicitId =
+ confParser.stringConf().option(SparkWriteOptions.OUTPUT_SORT_ORDER_ID).parseOptional();
+
+ if (explicitId != null) {
+ int id = Integer.parseInt(explicitId);
+ Preconditions.checkArgument(
+ table.sortOrders().containsKey(id),
+ "Cannot use output sort order id %s because the table does not contain a sort order with that id",
+ id);
+ return id;
+ }
+
+ if (writeRequirements.hasOrdering()) {
+ return table.sortOrder().orderId();
Review Comment:
> Why are we using the table's sort order and not the ordering from the writeRequirements?
There's lots of great candor outlining this above, but essentially iceberg sort order != spark sort order -> why this PR has a bit of complexity.
The logic performed above essentially is saying:
- if a sort order was explicitly pinned during a write w/ OUTPUT_SORT_ORDER (set by rewrite data files)
- else, if spark has a sort order set, it _must_ correspond to the iceberg sort order
I do agree that it's a bit clever and I had originally opted for a bit more of an explicit approach as agreed upon with the maintainers above.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org