You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "dramaticlly (via GitHub)" <gi...@apache.org> on 2024/08/29 22:57:13 UTC
[PR] Spark 3.5: Skip sort for incomparable data types in CreateChangelogViewProcedure [iceberg]
dramaticlly opened a new pull request, #11045:
URL: https://github.com/apache/iceberg/pull/11045
This help fix a bug we observed in CreateChangelogViewProcedure when base table contain incomparable columns such as map.
```
org.apache.spark.sql.AnalysisException: [DATATYPE_MISMATCH.INVALID_ORDERING_TYPE] Cannot resolve "tokens ASC NULLS FIRST" due to data type mismatch: The `sortorder` does not support ordering on type "MAP<STRING, STRING>"
org.apache.spark.sql.catalyst.analysis.package$AnalysisErrorAt.dataTypeMismatch(package.scala:73)
org.apache.spark.sql.catalyst.analysis.CheckAnalysis.$anonfun$checkAnalysis0$5(CheckAnalysis.scala:269)
org.apache.spark.sql.catalyst.analysis.CheckAnalysis.$anonfun$checkAnalysis0$5$adapted(CheckAnalysis.scala:256)
org.apache.spark.sql.catalyst.trees.TreeNode.foreachUp(TreeNode.scala:295)
org.apache.spark.sql.catalyst.analysis.CheckAnalysis.$anonfun$checkAnalysis0$4(CheckAnalysis.scala:256)
org.apache.spark.sql.catalyst.analysis.CheckAnalysis.$anonfun$checkAnalysis0$4$adapted(CheckAnalysis.scala:256)
scala.collection.immutable.Stream.foreach(Stream.scala:533)
org.apache.spark.sql.catalyst.analysis.CheckAnalysis.$anonfun$checkAnalysis0$1(CheckAnalysis.scala:256)
org.apache.spark.sql.catalyst.analysis.CheckAnalysis.$anonfun$checkAnalysis0$1$adapted(CheckAnalysis.scala:163)
org.apache.spark.sql.catalyst.trees.TreeNode.foreachUp(TreeNode.scala:295)
org.apache.spark.sql.catalyst.analysis.CheckAnalysis.checkAnalysis0(CheckAnalysis.scala:163)
org.apache.spark.sql.catalyst.analysis.CheckAnalysis.checkAnalysis0$(CheckAnalysis.scala:160)
org.apache.spark.sql.catalyst.analysis.Analyzer.checkAnalysis0(Analyzer.scala:188)
org.apache.spark.sql.catalyst.analysis.CheckAnalysis.checkAnalysis(CheckAnalysis.scala:156)
org.apache.spark.sql.catalyst.analysis.CheckAnalysis.checkAnalysis$(CheckAnalysis.scala:146)
org.apache.spark.sql.catalyst.analysis.Analyzer.checkAnalysis(Analyzer.scala:188)
org.apache.spark.sql.catalyst.analysis.Analyzer.$anonfun$executeAndCheck$1(Analyzer.scala:211)
org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper$.markInAnalyzer(AnalysisHelper.scala:330)
org.apache.spark.sql.catalyst.analysis.Analyzer.executeAndCheck(Analyzer.scala:208)
org.apache.spark.sql.execution.QueryExecution.$anonfun$analyzed$1(QueryExecution.scala:78)
org.apache.spark.sql.catalyst.QueryPlanningTracker.measurePhase(QueryPlanningTracker.scala:111)
org.apache.spark.sql.execution.QueryExecution.$anonfun$executePhase$2(QueryExecution.scala:204)
org.apache.spark.sql.execution.QueryExecution$.withInternalError(QueryExecution.scala:547)
org.apache.spark.sql.execution.QueryExecution.$anonfun$executePhase$1(QueryExecution.scala:204)
org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:827)
org.apache.spark.sql.execution.QueryExecution.executePhase(QueryExecution.scala:203)
org.apache.spark.sql.execution.QueryExecution.analyzed$lzycompute(QueryExecution.scala:78)
org.apache.spark.sql.execution.QueryExecution.analyzed(QueryExecution.scala:76)
org.apache.spark.sql.execution.QueryExecution.assertAnalyzed(QueryExecution.scala:68)
org.apache.spark.sql.Dataset.<init>(Dataset.scala:204)
org.apache.spark.sql.Dataset.<init>(Dataset.scala:210)
org.apache.spark.sql.Dataset$.apply(Dataset.scala:74)
org.apache.spark.sql.Dataset.withTypedPlan(Dataset.scala:4242)
org.apache.spark.sql.Dataset.sortInternal(Dataset.scala:4230)
org.apache.spark.sql.Dataset.sortWithinPartitions(Dataset.scala:1381)
org.apache.spark.sql.Dataset.sortWithinPartitions(Dataset.scala:1380)
org.apache.iceberg.spark.procedures.CreateChangelogViewProcedure.applyCarryoverRemoveIterator(CreateChangelogViewProcedure.java:267)
org.apache.iceberg.spark.procedures.CreateChangelogViewProcedure.removeCarryoverRows(CreateChangelogViewProcedure.java:218)
org.apache.iceberg.spark.procedures.CreateChangelogViewProcedure.call(CreateChangelogViewProcedure.java:167)
```
This is due to we passed incomparable columns to spark using sortWithinPartitions in [here](https://github.com/apache/iceberg/blob/main/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java#L236).
Find some old spark PR where it consider both maps and binary data type are not comparable/sortable in [SPARK-9415 ](https://issues.apache.org/jira/browse/SPARK-9415) and [code](https://github.com/apache/spark/blob/515dfc166a95ecc8decce0f0cd99e06fe395f94f/sql/api/src/main/scala/org/apache/spark/sql/catalyst/expressions/OrderUtils.scala#L25-L32)
--
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 3.5: Skip sort for incomparable data types in CreateChangelogViewProcedure [iceberg]
Posted by "karuppayya (via GitHub)" <gi...@apache.org>.
karuppayya commented on code in PR #11045:
URL: https://github.com/apache/iceberg/pull/11045#discussion_r1739341390
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java:
##########
@@ -242,6 +249,34 @@ public void testUpdateWithIdentifierField() {
sql("select * from %s order by _change_ordinal, id, data", viewName));
}
+ @TestTemplate
+ public void testUpdateWithInComparableType() {
+ createTableWithIncomparableType();
+
+ sql("INSERT INTO %s VALUES (2, Map('b','b'), 12)", tableName);
+ Table table = validationCatalog.loadTable(tableIdent);
+ Snapshot snap1 = table.currentSnapshot();
+
+ sql("INSERT OVERWRITE %s VALUES (3, Map('c','c'), 13), (2, Map('d','d'), 12)", tableName);
+ table.refresh();
+ Snapshot snap2 = table.currentSnapshot();
+
+ assertThat(table.sortOrder()).isNotNull();
+
+ List<Object[]> returns =
+ sql("CALL %s.system.create_changelog_view(table => '%s')", catalogName, tableName);
+
+ String viewName = (String) returns.get(0)[0];
+ assertEquals(
+ "Rows should match",
+ ImmutableList.of(
+ row(2, Collections.singletonMap("b", "b"), 12, INSERT, 0, snap1.snapshotId()),
+ row(2, Collections.singletonMap("b", "b"), 12, DELETE, 1, snap2.snapshotId()),
+ row(2, Collections.singletonMap("d", "d"), 12, INSERT, 1, snap2.snapshotId()),
+ row(3, Collections.singletonMap("c", "c"), 13, INSERT, 1, snap2.snapshotId())),
+ sql("select * from %s order by _change_ordinal, id", viewName));
Review Comment:
yeah,
looked like a standalone sql on first look :) .
--
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 3.5: Mandate identifier fields when create_changelog_view for table contain unsortable columns [iceberg]
Posted by "flyrain (via GitHub)" <gi...@apache.org>.
flyrain merged PR #11045:
URL: https://github.com/apache/iceberg/pull/11045
--
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 3.5: Mandate identifier fields when create_changelog_view for table contain unsortable columns [iceberg]
Posted by "dramaticlly (via GitHub)" <gi...@apache.org>.
dramaticlly commented on code in PR #11045:
URL: https://github.com/apache/iceberg/pull/11045#discussion_r1744650323
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java:
##########
@@ -146,10 +147,16 @@ public InternalRow[] call(InternalRow args) {
Dataset<Row> df = loadRows(changelogTableIdent, options(input));
boolean netChanges = input.asBoolean(NET_CHANGES, false);
+ String[] identifierColumns = identifierColumns(input, tableIdent);
+ Preconditions.checkArgument(
+ identifierColumns.length > 0
+ || Arrays.stream(df.schema().fields())
+ .allMatch(field -> OrderUtils.isOrderable(field.dataType())),
+ "Identifier field is required if table contains unorderable columns");
Review Comment:
I agree it would be informative to know which column, but I feel the desired result would be add identifier column to the table or procedure instead to uniquely identify the row instead of drop the column/update the type.
--
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 3.5: Skip sort for incomparable data types in CreateChangelogViewProcedure [iceberg]
Posted by "karuppayya (via GitHub)" <gi...@apache.org>.
karuppayya commented on code in PR #11045:
URL: https://github.com/apache/iceberg/pull/11045#discussion_r1739438634
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java:
##########
@@ -242,6 +249,34 @@ public void testUpdateWithIdentifierField() {
sql("select * from %s order by _change_ordinal, id, data", viewName));
}
+ @TestTemplate
+ public void testUpdateWithInComparableType() {
+ createTableWithIncomparableType();
+
+ sql("INSERT INTO %s VALUES (2, Map('b','b'), 12)", tableName);
+ Table table = validationCatalog.loadTable(tableIdent);
+ Snapshot snap1 = table.currentSnapshot();
+
+ sql("INSERT OVERWRITE %s VALUES (3, Map('c','c'), 13), (2, Map('d','d'), 12)", tableName);
Review Comment:
I can see code where [we compare every column of two rows](https://github.com/apache/iceberg/blob/main/spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/ChangelogIterator.java#L101) in a iterator to see if they are equal and base that to generate the change log.
For example consider a table with schema <id: Int, map: Map<String, String>> having records
` 1, Map<a, a>`, `1, Map<b, b>`, `1, Map<a, a>`
An ordering based on `id`
can result in in
```
1, Map<a, a>
1, Map<a, a>
1, Map<b, b>
```
In this case first two rows are same
or
```
1, Map<a, a>
1, Map<b, b>
1, Map<a, a>
```
In this case first two rows are different.
Hence I was curious if the result would differ when we skip the ordering on a column.
--
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 3.5: Mandate identifier fields when create_changelog_view for table contain unsortable columns [iceberg]
Posted by "flyrain (via GitHub)" <gi...@apache.org>.
flyrain commented on code in PR #11045:
URL: https://github.com/apache/iceberg/pull/11045#discussion_r1745879562
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java:
##########
@@ -146,10 +147,16 @@ public InternalRow[] call(InternalRow args) {
Dataset<Row> df = loadRows(changelogTableIdent, options(input));
boolean netChanges = input.asBoolean(NET_CHANGES, false);
+ String[] identifierColumns = identifierColumns(input, tableIdent);
+ Preconditions.checkArgument(
+ identifierColumns.length > 0
+ || Arrays.stream(df.schema().fields())
+ .allMatch(field -> OrderUtils.isOrderable(field.dataType())),
+ "Identifier field is required if table contains unorderable columns");
Review Comment:
I have to agree with @karuppayya that, the error message now is more vague than before
```
The `sortorder` does not support ordering on type "MAP<STRING, STRING>"
```
I got that we want user to specify the identifier columns, but it's still valuable to provide more error details, so that users can make the choice by themselves.
--
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 3.5: Skip sort for incomparable data types in CreateChangelogViewProcedure [iceberg]
Posted by "anigos (via GitHub)" <gi...@apache.org>.
anigos commented on code in PR #11045:
URL: https://github.com/apache/iceberg/pull/11045#discussion_r1739596762
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java:
##########
@@ -242,6 +249,34 @@ public void testUpdateWithIdentifierField() {
sql("select * from %s order by _change_ordinal, id, data", viewName));
}
+ @TestTemplate
+ public void testUpdateWithInComparableType() {
+ createTableWithIncomparableType();
+
+ sql("INSERT INTO %s VALUES (2, Map('b','b'), 12)", tableName);
+ Table table = validationCatalog.loadTable(tableIdent);
+ Snapshot snap1 = table.currentSnapshot();
+
+ sql("INSERT OVERWRITE %s VALUES (3, Map('c','c'), 13), (2, Map('d','d'), 12)", tableName);
Review Comment:
Exactly what I thought. We can't skip the ordering for a specific column because any non key column are under DML consideration. It can impact the net changes outcome. @flyrain what you say?
--
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] Doc: Add warning for create_changelog_view when columns are unorderable [iceberg]
Posted by "karuppayya (via GitHub)" <gi...@apache.org>.
karuppayya commented on PR #11045:
URL: https://github.com/apache/iceberg/pull/11045#issuecomment-2329789217
@dramaticlly Should we also validate the schema for un-orderable in the procedure and throw a `org.apache.iceberg.exceptions.ValidationException`?
--
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 3.5: Mandate identifier fields when create_changelog_view for table contain unsortable columns [iceberg]
Posted by "dramaticlly (via GitHub)" <gi...@apache.org>.
dramaticlly commented on code in PR #11045:
URL: https://github.com/apache/iceberg/pull/11045#discussion_r1745898693
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java:
##########
@@ -146,10 +147,16 @@ public InternalRow[] call(InternalRow args) {
Dataset<Row> df = loadRows(changelogTableIdent, options(input));
boolean netChanges = input.asBoolean(NET_CHANGES, false);
+ String[] identifierColumns = identifierColumns(input, tableIdent);
+ Preconditions.checkArgument(
+ identifierColumns.length > 0
+ || Arrays.stream(df.schema().fields())
+ .allMatch(field -> OrderUtils.isOrderable(field.dataType())),
+ "Identifier field is required if table contains unorderable columns");
Review Comment:
thank you @flyrain and @karuppayya and that's a good point to judge based on existing error message , added unorderable column name as suggested.
--
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 3.5: Skip sort for incomparable data types in CreateChangelogViewProcedure [iceberg]
Posted by "dramaticlly (via GitHub)" <gi...@apache.org>.
dramaticlly commented on code in PR #11045:
URL: https://github.com/apache/iceberg/pull/11045#discussion_r1739209075
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java:
##########
@@ -242,6 +249,34 @@ public void testUpdateWithIdentifierField() {
sql("select * from %s order by _change_ordinal, id, data", viewName));
}
+ @TestTemplate
+ public void testUpdateWithInComparableType() {
+ createTableWithIncomparableType();
+
+ sql("INSERT INTO %s VALUES (2, Map('b','b'), 12)", tableName);
+ Table table = validationCatalog.loadTable(tableIdent);
+ Snapshot snap1 = table.currentSnapshot();
+
+ sql("INSERT OVERWRITE %s VALUES (3, Map('c','c'), 13), (2, Map('d','d'), 12)", tableName);
Review Comment:
Let me trying to think out loud
- Given the identifier column is missing, we are basically sort all columns to generate changelog, here we are skip sorting on the columns which cannot be ordered. I dont feel like we will lose on detect the change as we did not filter any row out in the results
- If identifier columns are added to the table or provided in procedure explicitly, [the repartition spec will only include such instead of all columns](https://github.com/apache/iceberg/blob/main/spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java#L164-L176) and later used for sorting
--
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 3.5: Mandate identifier fields when create_changelog_view for table contain unsortable columns [iceberg]
Posted by "dramaticlly (via GitHub)" <gi...@apache.org>.
dramaticlly commented on code in PR #11045:
URL: https://github.com/apache/iceberg/pull/11045#discussion_r1744650323
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java:
##########
@@ -146,10 +147,16 @@ public InternalRow[] call(InternalRow args) {
Dataset<Row> df = loadRows(changelogTableIdent, options(input));
boolean netChanges = input.asBoolean(NET_CHANGES, false);
+ String[] identifierColumns = identifierColumns(input, tableIdent);
+ Preconditions.checkArgument(
+ identifierColumns.length > 0
+ || Arrays.stream(df.schema().fields())
+ .allMatch(field -> OrderUtils.isOrderable(field.dataType())),
+ "Identifier field is required if table contains unorderable columns");
Review Comment:
I agree it would be informative to know which column, but I feel the desired result would be add identifier column to the table or procedure to uniquely identify the row instead of drop the column/update the type.
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java:
##########
@@ -146,10 +147,16 @@ public InternalRow[] call(InternalRow args) {
Dataset<Row> df = loadRows(changelogTableIdent, options(input));
boolean netChanges = input.asBoolean(NET_CHANGES, false);
+ String[] identifierColumns = identifierColumns(input, tableIdent);
+ Preconditions.checkArgument(
+ identifierColumns.length > 0
+ || Arrays.stream(df.schema().fields())
+ .allMatch(field -> OrderUtils.isOrderable(field.dataType())),
+ "Identifier field is required if table contains unorderable columns");
Review Comment:
I agree it would be informative to know which column, but I feel the desired result would be add identifier column to the table or procedure to uniquely identify the row instead of drop the unorderable column/update the type.
--
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 3.5: Skip sort for incomparable data types in CreateChangelogViewProcedure [iceberg]
Posted by "flyrain (via GitHub)" <gi...@apache.org>.
flyrain commented on code in PR #11045:
URL: https://github.com/apache/iceberg/pull/11045#discussion_r1739888471
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java:
##########
@@ -183,21 +185,26 @@ private boolean shouldComputeUpdateImages(ProcedureInput input) {
}
private Dataset<Row> removeCarryoverRows(Dataset<Row> df, boolean netChanges) {
- Predicate<String> columnsToKeep;
- if (netChanges) {
- Set<String> metadataColumn =
- Sets.newHashSet(
- MetadataColumns.CHANGE_TYPE.name(),
- MetadataColumns.CHANGE_ORDINAL.name(),
- MetadataColumns.COMMIT_SNAPSHOT_ID.name());
-
- columnsToKeep = column -> !metadataColumn.contains(column);
- } else {
- columnsToKeep = column -> !column.equals(MetadataColumns.CHANGE_TYPE.name());
- }
+ Set<String> metadataColumn =
+ netChanges
+ ? Sets.newHashSet(
+ MetadataColumns.CHANGE_TYPE.name(),
+ MetadataColumns.CHANGE_ORDINAL.name(),
+ MetadataColumns.COMMIT_SNAPSHOT_ID.name())
+ : Sets.newHashSet(MetadataColumns.CHANGE_TYPE.name());
+
+ Predicate<StructField> columnsToDiscard =
+ field ->
+ metadataColumn.contains(field.name())
+ // avoid sort on incomparable columns
+ || field.dataType() instanceof MapType
+ || field.dataType() instanceof BinaryType;
Column[] repartitionSpec =
- Arrays.stream(df.columns()).filter(columnsToKeep).map(df::col).toArray(Column[]::new);
+ Arrays.stream(df.schema().fields())
+ .filter(Predicate.not(columnsToDiscard))
Review Comment:
This breaks the [assumption](https://github.com/apache/iceberg/blob/79620e198009fa243c278c66fd442d107b46206a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/RemoveCarryoverIterator.java#L28-L28) of `RemoveCarryoverIterator`. For example, this is an order can happen now, but it will break `RemoveCarryoverIterator`, as we didn't remove the carryover rows, row1 and row3.
```
row1: (id=1, map=('a', 'a'), op='DELETE')
row2: (id=1, map=('a', 'b')', op='INSERT')
row3: (id=1, map=('a', 'a'), op='INSERT')
```
--
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 3.5: Skip sort for incomparable data types in CreateChangelogViewProcedure [iceberg]
Posted by "flyrain (via GitHub)" <gi...@apache.org>.
flyrain commented on code in PR #11045:
URL: https://github.com/apache/iceberg/pull/11045#discussion_r1739882108
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java:
##########
@@ -242,6 +249,34 @@ public void testUpdateWithIdentifierField() {
sql("select * from %s order by _change_ordinal, id, data", viewName));
}
+ @TestTemplate
+ public void testUpdateWithInComparableType() {
+ createTableWithIncomparableType();
+
+ sql("INSERT INTO %s VALUES (2, Map('b','b'), 12)", tableName);
+ Table table = validationCatalog.loadTable(tableIdent);
+ Snapshot snap1 = table.currentSnapshot();
+
+ sql("INSERT OVERWRITE %s VALUES (3, Map('c','c'), 13), (2, Map('d','d'), 12)", tableName);
Review Comment:
That's valid concern. Skipping certain columns breaks the [assumption](https://github.com/apache/iceberg/blob/79620e198009fa243c278c66fd442d107b46206a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/RemoveCarryoverIterator.java#L28-L28) of `ChangelogIterator`. I'd suggest to start from the class `ChangelogIterator` to see how it works with dataset having certain columns not sorted.
--
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 3.5: Skip sort for incomparable data types in CreateChangelogViewProcedure [iceberg]
Posted by "karuppayya (via GitHub)" <gi...@apache.org>.
karuppayya commented on code in PR #11045:
URL: https://github.com/apache/iceberg/pull/11045#discussion_r1737628804
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java:
##########
@@ -242,6 +249,34 @@ public void testUpdateWithIdentifierField() {
sql("select * from %s order by _change_ordinal, id, data", viewName));
}
+ @TestTemplate
+ public void testUpdateWithInComparableType() {
+ createTableWithIncomparableType();
+
+ sql("INSERT INTO %s VALUES (2, Map('b','b'), 12)", tableName);
+ Table table = validationCatalog.loadTable(tableIdent);
+ Snapshot snap1 = table.currentSnapshot();
+
+ sql("INSERT OVERWRITE %s VALUES (3, Map('c','c'), 13), (2, Map('d','d'), 12)", tableName);
+ table.refresh();
+ Snapshot snap2 = table.currentSnapshot();
+
+ assertThat(table.sortOrder()).isNotNull();
+
+ List<Object[]> returns =
+ sql("CALL %s.system.create_changelog_view(table => '%s')", catalogName, tableName);
+
+ String viewName = (String) returns.get(0)[0];
+ assertEquals(
+ "Rows should match",
+ ImmutableList.of(
+ row(2, Collections.singletonMap("b", "b"), 12, INSERT, 0, snap1.snapshotId()),
+ row(2, Collections.singletonMap("b", "b"), 12, DELETE, 1, snap2.snapshotId()),
+ row(2, Collections.singletonMap("d", "d"), 12, INSERT, 1, snap2.snapshotId()),
+ row(3, Collections.singletonMap("c", "c"), 13, INSERT, 1, snap2.snapshotId())),
+ sql("select * from %s order by _change_ordinal, id", viewName));
Review Comment:
Should we remove this?
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java:
##########
@@ -183,21 +185,26 @@ private boolean shouldComputeUpdateImages(ProcedureInput input) {
}
private Dataset<Row> removeCarryoverRows(Dataset<Row> df, boolean netChanges) {
- Predicate<String> columnsToKeep;
- if (netChanges) {
- Set<String> metadataColumn =
- Sets.newHashSet(
- MetadataColumns.CHANGE_TYPE.name(),
- MetadataColumns.CHANGE_ORDINAL.name(),
- MetadataColumns.COMMIT_SNAPSHOT_ID.name());
-
- columnsToKeep = column -> !metadataColumn.contains(column);
- } else {
- columnsToKeep = column -> !column.equals(MetadataColumns.CHANGE_TYPE.name());
- }
+ Set<String> metadataColumn =
+ netChanges
+ ? Sets.newHashSet(
+ MetadataColumns.CHANGE_TYPE.name(),
+ MetadataColumns.CHANGE_ORDINAL.name(),
+ MetadataColumns.COMMIT_SNAPSHOT_ID.name())
+ : Sets.newHashSet(MetadataColumns.CHANGE_TYPE.name());
+
+ Predicate<StructField> columnsToDiscard =
+ field ->
+ metadataColumn.contains(field.name())
+ // avoid sort on incomparable columns
+ || field.dataType() instanceof MapType
Review Comment:
Like mentioned in [PR desc](https://github.com/apache/spark/blob/master/sql/api/src/main/scala/org/apache/spark/sql/catalyst/expressions/OrderUtils.scala#L25), should we skip all the non-orderable types
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java:
##########
@@ -242,6 +249,34 @@ public void testUpdateWithIdentifierField() {
sql("select * from %s order by _change_ordinal, id, data", viewName));
}
+ @TestTemplate
+ public void testUpdateWithInComparableType() {
+ createTableWithIncomparableType();
+
+ sql("INSERT INTO %s VALUES (2, Map('b','b'), 12)", tableName);
+ Table table = validationCatalog.loadTable(tableIdent);
+ Snapshot snap1 = table.currentSnapshot();
+
+ sql("INSERT OVERWRITE %s VALUES (3, Map('c','c'), 13), (2, Map('d','d'), 12)", tableName);
Review Comment:
Doesn't generating change logs rows depend on the ordering of the rows?
If the ordering is removed on specific column, it can potentially return incorrect results?
For example if the first column type was Map, we would sort on the remaining columns. And we might not be able to detect chnages reliably?
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java:
##########
@@ -242,6 +249,34 @@ public void testUpdateWithIdentifierField() {
sql("select * from %s order by _change_ordinal, id, data", viewName));
}
+ @TestTemplate
+ public void testUpdateWithInComparableType() {
+ createTableWithIncomparableType();
Review Comment:
sql can be inlined?
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java:
##########
@@ -183,21 +185,26 @@ private boolean shouldComputeUpdateImages(ProcedureInput input) {
}
private Dataset<Row> removeCarryoverRows(Dataset<Row> df, boolean netChanges) {
- Predicate<String> columnsToKeep;
- if (netChanges) {
- Set<String> metadataColumn =
- Sets.newHashSet(
- MetadataColumns.CHANGE_TYPE.name(),
- MetadataColumns.CHANGE_ORDINAL.name(),
- MetadataColumns.COMMIT_SNAPSHOT_ID.name());
-
- columnsToKeep = column -> !metadataColumn.contains(column);
- } else {
- columnsToKeep = column -> !column.equals(MetadataColumns.CHANGE_TYPE.name());
- }
+ Set<String> metadataColumn =
+ netChanges
+ ? Sets.newHashSet(
+ MetadataColumns.CHANGE_TYPE.name(),
+ MetadataColumns.CHANGE_ORDINAL.name(),
+ MetadataColumns.COMMIT_SNAPSHOT_ID.name())
+ : Sets.newHashSet(MetadataColumns.CHANGE_TYPE.name());
+
+ Predicate<StructField> columnsToDiscard =
+ field ->
+ metadataColumn.contains(field.name())
+ // avoid sort on incomparable columns
+ || field.dataType() instanceof MapType
+ || field.dataType() instanceof BinaryType;
Column[] repartitionSpec =
- Arrays.stream(df.columns()).filter(columnsToKeep).map(df::col).toArray(Column[]::new);
+ Arrays.stream(df.schema().fields())
+ .filter(Predicate.not(columnsToDiscard))
Review Comment:
Looks like we use it in the repartitioning spec as well, do we need to remove from the columns to be repartitioned 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 3.5: Skip sort for incomparable data types in CreateChangelogViewProcedure [iceberg]
Posted by "flyrain (via GitHub)" <gi...@apache.org>.
flyrain commented on code in PR #11045:
URL: https://github.com/apache/iceberg/pull/11045#discussion_r1739891387
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java:
##########
@@ -183,21 +185,26 @@ private boolean shouldComputeUpdateImages(ProcedureInput input) {
}
private Dataset<Row> removeCarryoverRows(Dataset<Row> df, boolean netChanges) {
- Predicate<String> columnsToKeep;
- if (netChanges) {
- Set<String> metadataColumn =
- Sets.newHashSet(
- MetadataColumns.CHANGE_TYPE.name(),
- MetadataColumns.CHANGE_ORDINAL.name(),
- MetadataColumns.COMMIT_SNAPSHOT_ID.name());
-
- columnsToKeep = column -> !metadataColumn.contains(column);
- } else {
- columnsToKeep = column -> !column.equals(MetadataColumns.CHANGE_TYPE.name());
- }
+ Set<String> metadataColumn =
+ netChanges
+ ? Sets.newHashSet(
+ MetadataColumns.CHANGE_TYPE.name(),
+ MetadataColumns.CHANGE_ORDINAL.name(),
+ MetadataColumns.COMMIT_SNAPSHOT_ID.name())
+ : Sets.newHashSet(MetadataColumns.CHANGE_TYPE.name());
+
+ Predicate<StructField> columnsToDiscard =
+ field ->
+ metadataColumn.contains(field.name())
+ // avoid sort on incomparable columns
+ || field.dataType() instanceof MapType
+ || field.dataType() instanceof BinaryType;
Column[] repartitionSpec =
- Arrays.stream(df.columns()).filter(columnsToKeep).map(df::col).toArray(Column[]::new);
+ Arrays.stream(df.schema().fields())
+ .filter(Predicate.not(columnsToDiscard))
Review Comment:
Agreed with @karuppayya that if Spark doesn't complain the repartition spec with map columns, we should keep them, so that we at least narrow down the rows within the same partition with all columns checked. We can skip them in `sortSpec`. However, that's not enough, we will need to make changes in `ChangeIterator` to make sure skipping-sorting-column works. WDYT?
--
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] Doc: Add warning for create_changelog_view when columns are unorderable [iceberg]
Posted by "sfc-gh-ygu (via GitHub)" <gi...@apache.org>.
sfc-gh-ygu commented on code in PR #11045:
URL: https://github.com/apache/iceberg/pull/11045#discussion_r1744280101
##########
docs/docs/spark-procedures.md:
##########
@@ -769,6 +769,10 @@ Here is a list of commonly used Spark read options:
* `start-timestamp`: the exclusive start timestamp. If not provided, it reads from the table’s first snapshot inclusively.
* `end-timestamp`: the inclusive end timestamp, default to table's current snapshot.
+!!! warning
+ `create_changelog_view` procedure rely on partitioning and sorting of all columns to identify carry-over rows when `identifer_columns` are not provided.
+ Keep in mind if table contains unorderable column (like Map or Binary type) and not configured with `identifer_columns`, this procedure will fail with AnalysisException.
Review Comment:
Just curious, binary type isn't a atomic type?https://github.com/apache/spark/blob/2ed6c3e511f322c5fd01953736c376a85ff2c687/sql/api/src/main/scala/org/apache/spark/sql/catalyst/expressions/OrderUtils.scala#L27
--
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] Doc: Add warning for create_changelog_view when columns are unorderable [iceberg]
Posted by "dramaticlly (via GitHub)" <gi...@apache.org>.
dramaticlly commented on PR #11045:
URL: https://github.com/apache/iceberg/pull/11045#issuecomment-2329762019
Had some offline discussion with @flyrain and decided to update documentation instead of the original code change, given the [row lineage proposal in v3](https://docs.google.com/document/d/146YuAnU17prnIhyuvbCtCtVSavyd5N7hKryyVRaFDTE/edit#heading=h.8vlydu9k3h0e) looks more promising as it will add a marker to every row in an Iceberg table to indicate its origin and it's better suited for CDC usage here.
The current approach rely on partitioning and sorting to identify the carry over rows cannot afford to skip unorderable columns and keep them in memory to identify the carry over rows in iterators.
--
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 3.5: Mandate identifier fields when create_changelog_view for table contain unsortable columns [iceberg]
Posted by "karuppayya (via GitHub)" <gi...@apache.org>.
karuppayya commented on code in PR #11045:
URL: https://github.com/apache/iceberg/pull/11045#discussion_r1744645623
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java:
##########
@@ -146,10 +147,16 @@ public InternalRow[] call(InternalRow args) {
Dataset<Row> df = loadRows(changelogTableIdent, options(input));
boolean netChanges = input.asBoolean(NET_CHANGES, false);
+ String[] identifierColumns = identifierColumns(input, tableIdent);
+ Preconditions.checkArgument(
+ identifierColumns.length > 0
+ || Arrays.stream(df.schema().fields())
+ .allMatch(field -> OrderUtils.isOrderable(field.dataType())),
+ "Identifier field is required if table contains unorderable columns");
Review Comment:
I think it would be informative if we can log the column name(and type) that is not order-able
--
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 3.5: Skip sort for incomparable data types in CreateChangelogViewProcedure [iceberg]
Posted by "karuppayya (via GitHub)" <gi...@apache.org>.
karuppayya commented on code in PR #11045:
URL: https://github.com/apache/iceberg/pull/11045#discussion_r1739438634
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java:
##########
@@ -242,6 +249,34 @@ public void testUpdateWithIdentifierField() {
sql("select * from %s order by _change_ordinal, id, data", viewName));
}
+ @TestTemplate
+ public void testUpdateWithInComparableType() {
+ createTableWithIncomparableType();
+
+ sql("INSERT INTO %s VALUES (2, Map('b','b'), 12)", tableName);
+ Table table = validationCatalog.loadTable(tableIdent);
+ Snapshot snap1 = table.currentSnapshot();
+
+ sql("INSERT OVERWRITE %s VALUES (3, Map('c','c'), 13), (2, Map('d','d'), 12)", tableName);
Review Comment:
I can see code where [we compare every column of two rows](https://github.com/apache/iceberg/blob/main/spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/ChangelogIterator.java#L101) in a iterator to see if they are equal and base that to generate the change log.
For example consider a table with schema <id: Int, map: Map<String, String>> having records
` 1, Map<a, a>`, `1, Map<b, b>`, `1, Map<a, a>`
An ordering based on `id`
can result in in
```
1, Map<a, a>
1, Map<a, a>
1, Map<b, b>
```
In this case first two rows are same
or
```
record1 => 1, Map<a, a>
record2 => 1, Map<b, b>
record3 => 1, Map<a, a>
```
In this case first two rows are different.
Hence I was curious if the result would differ when we skip the ordering on a column.
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java:
##########
@@ -183,21 +185,26 @@ private boolean shouldComputeUpdateImages(ProcedureInput input) {
}
private Dataset<Row> removeCarryoverRows(Dataset<Row> df, boolean netChanges) {
- Predicate<String> columnsToKeep;
- if (netChanges) {
- Set<String> metadataColumn =
- Sets.newHashSet(
- MetadataColumns.CHANGE_TYPE.name(),
- MetadataColumns.CHANGE_ORDINAL.name(),
- MetadataColumns.COMMIT_SNAPSHOT_ID.name());
-
- columnsToKeep = column -> !metadataColumn.contains(column);
- } else {
- columnsToKeep = column -> !column.equals(MetadataColumns.CHANGE_TYPE.name());
- }
+ Set<String> metadataColumn =
+ netChanges
+ ? Sets.newHashSet(
+ MetadataColumns.CHANGE_TYPE.name(),
+ MetadataColumns.CHANGE_ORDINAL.name(),
+ MetadataColumns.COMMIT_SNAPSHOT_ID.name())
+ : Sets.newHashSet(MetadataColumns.CHANGE_TYPE.name());
+
+ Predicate<StructField> columnsToDiscard =
+ field ->
+ metadataColumn.contains(field.name())
+ // avoid sort on incomparable columns
+ || field.dataType() instanceof MapType
+ || field.dataType() instanceof BinaryType;
Column[] repartitionSpec =
- Arrays.stream(df.columns()).filter(columnsToKeep).map(df::col).toArray(Column[]::new);
+ Arrays.stream(df.schema().fields())
+ .filter(Predicate.not(columnsToDiscard))
Review Comment:
I meant to say,
we are now removing the columns from repartitioning spec as well, which was not the case earlier. Is that intended?
--
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 3.5: Skip sort for incomparable data types in CreateChangelogViewProcedure [iceberg]
Posted by "flyrain (via GitHub)" <gi...@apache.org>.
flyrain commented on PR #11045:
URL: https://github.com/apache/iceberg/pull/11045#issuecomment-2323003542
Thanks @dramaticlly for the PR. We will need to look a bit deeper to see how it works. The `ChangeIterator` relies on the partition and sort. I didn't see a way to skip a column without changes from iterator. Admittedly, the iterator itself is a bit fragile, but it seems the most efficient way when we implement it. Hopefully the new row lineage spec in table spec V3 will help a lot 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 3.5: Mandate identifier fields when create_changelog_view for table contain unsortable columns [iceberg]
Posted by "flyrain (via GitHub)" <gi...@apache.org>.
flyrain commented on PR #11045:
URL: https://github.com/apache/iceberg/pull/11045#issuecomment-2332359477
Thanks @dramaticlly for the change, and thanks @karuppayya @anigos @huaxingao for the 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 3.5: Skip sort for incomparable data types in CreateChangelogViewProcedure [iceberg]
Posted by "dramaticlly (via GitHub)" <gi...@apache.org>.
dramaticlly commented on code in PR #11045:
URL: https://github.com/apache/iceberg/pull/11045#discussion_r1739201510
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java:
##########
@@ -242,6 +249,34 @@ public void testUpdateWithIdentifierField() {
sql("select * from %s order by _change_ordinal, id, data", viewName));
}
+ @TestTemplate
+ public void testUpdateWithInComparableType() {
+ createTableWithIncomparableType();
+
+ sql("INSERT INTO %s VALUES (2, Map('b','b'), 12)", tableName);
+ Table table = validationCatalog.loadTable(tableIdent);
+ Snapshot snap1 = table.currentSnapshot();
+
+ sql("INSERT OVERWRITE %s VALUES (3, Map('c','c'), 13), (2, Map('d','d'), 12)", tableName);
+ table.refresh();
+ Snapshot snap2 = table.currentSnapshot();
+
+ assertThat(table.sortOrder()).isNotNull();
+
+ List<Object[]> returns =
+ sql("CALL %s.system.create_changelog_view(table => '%s')", catalogName, tableName);
+
+ String viewName = (String) returns.get(0)[0];
+ assertEquals(
+ "Rows should match",
+ ImmutableList.of(
+ row(2, Collections.singletonMap("b", "b"), 12, INSERT, 0, snap1.snapshotId()),
+ row(2, Collections.singletonMap("b", "b"), 12, DELETE, 1, snap2.snapshotId()),
+ row(2, Collections.singletonMap("d", "d"), 12, INSERT, 1, snap2.snapshotId()),
+ row(3, Collections.singletonMap("c", "c"), 13, INSERT, 1, snap2.snapshotId())),
+ sql("select * from %s order by _change_ordinal, id", viewName));
Review Comment:
this might look a bit confusing, but the assert equals are asserting expectation vs reality, here we are building expected using immutable list and row and last line of sql is where we get the reality, so we cannot remove this
```
assertEquals(
// context or error message if assert fails
"Rows should match",
// expected
ImmutableList.of(
row(2, Collections.singletonMap("b", "b"), 12, INSERT, 0, snap1.snapshotId()),
row(2, Collections.singletonMap("b", "b"), 12, DELETE, 1, snap2.snapshotId()),
row(2, Collections.singletonMap("d", "d"), 12, INSERT, 1, snap2.snapshotId()),
row(3, Collections.singletonMap("c", "c"), 13, INSERT, 1, snap2.snapshotId())),
// actual
sql("select * from %s order by _change_ordinal, id", viewName));
```
--
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 3.5: Mandate identifier fields when create_changelog_view for table contain unsortable columns [iceberg]
Posted by "dramaticlly (via GitHub)" <gi...@apache.org>.
dramaticlly commented on code in PR #11045:
URL: https://github.com/apache/iceberg/pull/11045#discussion_r1744599722
##########
docs/docs/spark-procedures.md:
##########
@@ -769,6 +769,10 @@ Here is a list of commonly used Spark read options:
* `start-timestamp`: the exclusive start timestamp. If not provided, it reads from the table’s first snapshot inclusively.
* `end-timestamp`: the inclusive end timestamp, default to table's current snapshot.
+!!! warning
+ `create_changelog_view` procedure rely on partitioning and sorting of all columns to identify carry-over rows when `identifer_columns` are not provided.
+ Keep in mind if table contains unorderable column (like Map or Binary type) and not configured with `identifer_columns`, this procedure will fail with AnalysisException.
Review Comment:
sorry I just double checked and binary type can be ordered. I also moved to add precondition check instead of update doc 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 3.5: Skip sort for incomparable data types in CreateChangelogViewProcedure [iceberg]
Posted by "dramaticlly (via GitHub)" <gi...@apache.org>.
dramaticlly commented on PR #11045:
URL: https://github.com/apache/iceberg/pull/11045#issuecomment-2322078735
@flyrain @szehon-ho or @huaxingao If you want to take a look?
--
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 3.5: Skip sort for incomparable data types in CreateChangelogViewProcedure [iceberg]
Posted by "dramaticlly (via GitHub)" <gi...@apache.org>.
dramaticlly commented on code in PR #11045:
URL: https://github.com/apache/iceberg/pull/11045#discussion_r1739199528
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java:
##########
@@ -183,21 +185,26 @@ private boolean shouldComputeUpdateImages(ProcedureInput input) {
}
private Dataset<Row> removeCarryoverRows(Dataset<Row> df, boolean netChanges) {
- Predicate<String> columnsToKeep;
- if (netChanges) {
- Set<String> metadataColumn =
- Sets.newHashSet(
- MetadataColumns.CHANGE_TYPE.name(),
- MetadataColumns.CHANGE_ORDINAL.name(),
- MetadataColumns.COMMIT_SNAPSHOT_ID.name());
-
- columnsToKeep = column -> !metadataColumn.contains(column);
- } else {
- columnsToKeep = column -> !column.equals(MetadataColumns.CHANGE_TYPE.name());
- }
+ Set<String> metadataColumn =
+ netChanges
+ ? Sets.newHashSet(
+ MetadataColumns.CHANGE_TYPE.name(),
+ MetadataColumns.CHANGE_ORDINAL.name(),
+ MetadataColumns.COMMIT_SNAPSHOT_ID.name())
+ : Sets.newHashSet(MetadataColumns.CHANGE_TYPE.name());
+
+ Predicate<StructField> columnsToDiscard =
+ field ->
+ metadataColumn.contains(field.name())
+ // avoid sort on incomparable columns
+ || field.dataType() instanceof MapType
Review Comment:
yeah that's a good point, let me use `OrderUtils.isOrderable(field.dataType())` instead
--
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 3.5: Skip sort for incomparable data types in CreateChangelogViewProcedure [iceberg]
Posted by "dramaticlly (via GitHub)" <gi...@apache.org>.
dramaticlly commented on code in PR #11045:
URL: https://github.com/apache/iceberg/pull/11045#discussion_r1739202341
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java:
##########
@@ -183,21 +185,26 @@ private boolean shouldComputeUpdateImages(ProcedureInput input) {
}
private Dataset<Row> removeCarryoverRows(Dataset<Row> df, boolean netChanges) {
- Predicate<String> columnsToKeep;
- if (netChanges) {
- Set<String> metadataColumn =
- Sets.newHashSet(
- MetadataColumns.CHANGE_TYPE.name(),
- MetadataColumns.CHANGE_ORDINAL.name(),
- MetadataColumns.COMMIT_SNAPSHOT_ID.name());
-
- columnsToKeep = column -> !metadataColumn.contains(column);
- } else {
- columnsToKeep = column -> !column.equals(MetadataColumns.CHANGE_TYPE.name());
- }
+ Set<String> metadataColumn =
+ netChanges
+ ? Sets.newHashSet(
+ MetadataColumns.CHANGE_TYPE.name(),
+ MetadataColumns.CHANGE_ORDINAL.name(),
+ MetadataColumns.COMMIT_SNAPSHOT_ID.name())
+ : Sets.newHashSet(MetadataColumns.CHANGE_TYPE.name());
+
+ Predicate<StructField> columnsToDiscard =
+ field ->
+ metadataColumn.contains(field.name())
+ // avoid sort on incomparable columns
+ || field.dataType() instanceof MapType
+ || field.dataType() instanceof BinaryType;
Column[] repartitionSpec =
- Arrays.stream(df.columns()).filter(columnsToKeep).map(df::col).toArray(Column[]::new);
+ Arrays.stream(df.schema().fields())
+ .filter(Predicate.not(columnsToDiscard))
Review Comment:
I think spark does not seem to complain about have those columns in repartitioned spec, if you have more spark background can you share a bit more what's the best practise and whether we shall remove?
--
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 3.5: Skip sort for incomparable data types in CreateChangelogViewProcedure [iceberg]
Posted by "flyrain (via GitHub)" <gi...@apache.org>.
flyrain commented on code in PR #11045:
URL: https://github.com/apache/iceberg/pull/11045#discussion_r1739891387
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java:
##########
@@ -183,21 +185,26 @@ private boolean shouldComputeUpdateImages(ProcedureInput input) {
}
private Dataset<Row> removeCarryoverRows(Dataset<Row> df, boolean netChanges) {
- Predicate<String> columnsToKeep;
- if (netChanges) {
- Set<String> metadataColumn =
- Sets.newHashSet(
- MetadataColumns.CHANGE_TYPE.name(),
- MetadataColumns.CHANGE_ORDINAL.name(),
- MetadataColumns.COMMIT_SNAPSHOT_ID.name());
-
- columnsToKeep = column -> !metadataColumn.contains(column);
- } else {
- columnsToKeep = column -> !column.equals(MetadataColumns.CHANGE_TYPE.name());
- }
+ Set<String> metadataColumn =
+ netChanges
+ ? Sets.newHashSet(
+ MetadataColumns.CHANGE_TYPE.name(),
+ MetadataColumns.CHANGE_ORDINAL.name(),
+ MetadataColumns.COMMIT_SNAPSHOT_ID.name())
+ : Sets.newHashSet(MetadataColumns.CHANGE_TYPE.name());
+
+ Predicate<StructField> columnsToDiscard =
+ field ->
+ metadataColumn.contains(field.name())
+ // avoid sort on incomparable columns
+ || field.dataType() instanceof MapType
+ || field.dataType() instanceof BinaryType;
Column[] repartitionSpec =
- Arrays.stream(df.columns()).filter(columnsToKeep).map(df::col).toArray(Column[]::new);
+ Arrays.stream(df.schema().fields())
+ .filter(Predicate.not(columnsToDiscard))
Review Comment:
Agreed with @karuppayya that if Spark don't complain the repartition spec with map columns, we should keep them. We can skip them in `sortSpec`. However, that's not enough, we will need to make changes in `ChangeIterator` to make sure skipping-sorting-column works. WDYT?
--
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