You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "bharos (via GitHub)" <gi...@apache.org> on 2025/02/19 07:36:43 UTC
[PR] Spark: Add option to provide partition spec in ADD_FILES procedure [iceberg]
bharos opened a new pull request, #12327:
URL: https://github.com/apache/iceberg/pull/12327
Closes #12325
--
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: Infer partition spec in ADD_FILES procedure for FileTables than taking latest table spec [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12327:
URL: https://github.com/apache/iceberg/pull/12327#discussion_r1968550486
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkTableUtil.java:
##########
@@ -1091,21 +1094,9 @@ private ExecutorService getService() {
/**
* Returns the first partition spec in an IcebergTable that shares the same names and ordering as
- * the partition columns in a given Spark Table. Throws an error if not found
+ * the partition columns provided. Throws an error if not found
*/
- private static PartitionSpec findCompatibleSpec(
- Table icebergTable, SparkSession spark, String sparkTable) throws AnalysisException {
- List<String> parts = Lists.newArrayList(Splitter.on('.').limit(2).split(sparkTable));
- String db = parts.size() == 1 ? "default" : parts.get(0);
- String table = parts.get(parts.size() == 1 ? 0 : 1);
-
- List<String> sparkPartNames =
- spark.catalog().listColumns(db, table).collectAsList().stream()
- .filter(org.apache.spark.sql.catalog.Column::isPartition)
- .map(org.apache.spark.sql.catalog.Column::name)
- .map(name -> name.toLowerCase(Locale.ROOT))
- .collect(Collectors.toList());
-
+ public static PartitionSpec findCompatibleSpec(List<String> partitionNames, Table icebergTable) {
Review Comment:
Could we keep this package private? Just to avoid exposing another api to the general public unless we have a need to.
--
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: Add option to provide partition spec in ADD_FILES procedure [iceberg]
Posted by "bharos (via GitHub)" <gi...@apache.org>.
bharos commented on code in PR #12327:
URL: https://github.com/apache/iceberg/pull/12327#discussion_r1962463880
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/Spark3Util.java:
##########
@@ -858,6 +858,23 @@ public static String quotedFullIdentifier(String catalogName, Identifier identif
.quoted();
}
+ public static org.apache.spark.sql.execution.datasources.PartitionSpec getInferredSpec(
+ SparkSession spark, Path rootPath) {
+ FileStatusCache fileStatusCache = FileStatusCache.getOrCreate(spark);
+ InMemoryFileIndex fileIndex =
+ new InMemoryFileIndex(
Review Comment:
Add files will do this InMemoryFileIndex calculation twice (first it will do here to infer the spec with option.empty()) and later it will be called in getPartitions (see L902) to list the files. I assume since fileStatusCache is passed with same spark session it will not do recomputation?
Else I will calculate the InMemoryFileIndex in AddFilesProcedure and infer schema there, and pass it along to getPartitions()
--
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: Infer partition spec in ADD_FILES procedure for FileTables than taking latest table spec [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12327:
URL: https://github.com/apache/iceberg/pull/12327#discussion_r1968547373
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAddFilesProcedure.java:
##########
@@ -635,6 +635,67 @@ public void addFilteredPartitionsToPartitionedWithNullValueFilteringOnDept() {
sql("SELECT id, name, dept, subdept FROM %s ORDER BY id", tableName));
}
+ @TestTemplate
+ public void addFileTableOldSpecDataAfterPartitionSpecEvolved() {
+ createPartitionedFileTable("parquet");
+ createIcebergTable(
+ "id Integer, name String, dept String, subdept String",
+ "PARTITIONED BY (id, dept, subdept)");
+ sql("ALTER TABLE %s DROP PARTITION FIELD dept", tableName);
+ sql(
+ "ALTER TABLE %s DROP PARTITION FIELD subdept",
+ tableName); // This spec is matching with the input data which is partitioned just by "id"
+ sql("ALTER TABLE %s ADD PARTITION FIELD subdept", tableName);
+
+ if (formatVersion == 1) {
+ assertThatThrownBy(
Review Comment:
You could also just do an "assume(formatVersion).greaterThan(1)" but checking the error case is fine too
--
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: Infer partition spec in ADD_FILES procedure for FileTables than taking latest table spec [iceberg]
Posted by "bharos (via GitHub)" <gi...@apache.org>.
bharos commented on PR #12327:
URL: https://github.com/apache/iceberg/pull/12327#issuecomment-2699333702
Thanks @RussellSpitzer for the review and guidance to impement this 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: Infer partition spec in ADD_FILES procedure for FileTables than taking latest table spec [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12327:
URL: https://github.com/apache/iceberg/pull/12327#discussion_r1968552113
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkTableUtil.java:
##########
@@ -1124,7 +1115,67 @@ private static PartitionSpec findCompatibleSpec(
throw new IllegalArgumentException(
String.format(
"Cannot find a partition spec in Iceberg table %s that matches the partition"
- + " columns (%s) in Spark table %s",
- icebergTable, sparkPartNames, sparkTable));
+ + " columns (%s) in input table",
+ icebergTable, partitionNames));
+ }
+
+ /**
+ * Returns the first partition spec in an IcebergTable that shares the same names and ordering as
+ * the partition columns in a given Spark Table. Throws an error if not found
+ */
+ private static PartitionSpec findCompatibleSpec(
+ Table icebergTable, SparkSession spark, String sparkTable) throws AnalysisException {
+ List<String> parts = Lists.newArrayList(Splitter.on('.').limit(2).split(sparkTable));
+ String db = parts.size() == 1 ? "default" : parts.get(0);
+ String table = parts.get(parts.size() == 1 ? 0 : 1);
+
+ List<String> sparkPartNames =
+ spark.catalog().listColumns(db, table).collectAsList().stream()
+ .filter(org.apache.spark.sql.catalog.Column::isPartition)
+ .map(org.apache.spark.sql.catalog.Column::name)
+ .map(name -> name.toLowerCase(Locale.ROOT))
+ .collect(Collectors.toList());
+ return findCompatibleSpec(sparkPartNames, icebergTable);
+ }
+
+ public static void validatePartitionFilter(
Review Comment:
I would keep this package private 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: Infer partition spec in ADD_FILES procedure rather than taking latest table spec [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12327:
URL: https://github.com/apache/iceberg/pull/12327#discussion_r1962479942
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/AddFilesProcedure.java:
##########
@@ -265,13 +283,13 @@ public String description() {
return "AddFiles";
}
- private void validatePartitionSpec(Table table, Map<String, String> partitionFilter) {
+ private void validatePartitionFilter(Table table, Map<String, String> partitionFilter) {
List<PartitionField> partitionFields = table.spec().fields();
Set<String> partitionNames =
table.spec().fields().stream().map(PartitionField::name).collect(Collectors.toSet());
Review Comment:
This is now wrong, we need to use the "compatible partition spec" we discovered instead of the current spec
--
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: Infer partition spec in ADD_FILES procedure rather than taking latest table spec [iceberg]
Posted by "bharos (via GitHub)" <gi...@apache.org>.
bharos commented on PR #12327:
URL: https://github.com/apache/iceberg/pull/12327#issuecomment-2669928722
In this PR, I am not handling the void transforms (as seen in the test case) and assuming that the ordering is same as the inferred spec 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: Infer partition spec in ADD_FILES procedure for FileTables than taking latest table spec [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12327:
URL: https://github.com/apache/iceberg/pull/12327#discussion_r1968555311
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/AddFilesProcedure.java:
##########
@@ -265,55 +271,16 @@ public String description() {
return "AddFiles";
}
- private void validatePartitionSpec(Table table, Map<String, String> partitionFilter) {
- List<PartitionField> partitionFields = table.spec().fields();
- Set<String> partitionNames =
- table.spec().fields().stream().map(PartitionField::name).collect(Collectors.toSet());
-
- boolean tablePartitioned = !partitionFields.isEmpty();
- boolean partitionSpecPassed = !partitionFilter.isEmpty();
-
+ private void nonIdentityPartitionCheck(Table table) {
Review Comment:
This is impossible now right? We've removed the ability to even locate a spec that isn't all identity columns
--
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: Add option to provide partition spec in ADD_FILES procedure [iceberg]
Posted by "bharos (via GitHub)" <gi...@apache.org>.
bharos commented on code in PR #12327:
URL: https://github.com/apache/iceberg/pull/12327#discussion_r1962464380
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAddFilesProcedure.java:
##########
@@ -607,6 +607,44 @@ public void addFilteredPartitionsToPartitionedWithNullValueFilteringOnDept() {
sql("SELECT id, name, dept, subdept FROM %s ORDER BY id", tableName));
}
+ @TestTemplate
+ public void addFileTableOldSpecDataAfterPartitionSpecEvolved() {
+ createPartitionedFileTable("parquet");
+ createIcebergTable(
+ "id Integer, name String, dept String, subdept String",
+ "PARTITIONED BY (id, dept, subdept)");
+ sql("ALTER TABLE %s DROP PARTITION FIELD dept", tableName);
+ sql(
+ "ALTER TABLE %s DROP PARTITION FIELD subdept",
+ tableName); // This spec is matching with the input data which is partitioned just by "id"
+ sql("ALTER TABLE %s ADD PARTITION FIELD subdept", tableName);
+
+ if (formatVersion == 1) {
+ assertThatThrownBy(
+ () ->
+ scalarSql(
+ "CALL %s.system.add_files('%s', '`parquet`.`%s`')",
Review Comment:
Note that this example drops and adds partition fields which creates void transforms in V1, so checking that case separately
--
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: Infer partition spec in ADD_FILES procedure for FileTables than taking latest table spec [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on PR #12327:
URL: https://github.com/apache/iceberg/pull/12327#issuecomment-2699316005
Thanks @bharos for 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: Infer partition spec in ADD_FILES procedure for FileTables than taking latest table spec [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12327:
URL: https://github.com/apache/iceberg/pull/12327#discussion_r1968545924
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAddFilesProcedure.java:
##########
@@ -635,6 +635,67 @@ public void addFilteredPartitionsToPartitionedWithNullValueFilteringOnDept() {
sql("SELECT id, name, dept, subdept FROM %s ORDER BY id", tableName));
}
+ @TestTemplate
+ public void addFileTableOldSpecDataAfterPartitionSpecEvolved() {
+ createPartitionedFileTable("parquet");
+ createIcebergTable(
+ "id Integer, name String, dept String, subdept String",
+ "PARTITIONED BY (id, dept, subdept)");
+ sql("ALTER TABLE %s DROP PARTITION FIELD dept", tableName);
+ sql(
+ "ALTER TABLE %s DROP PARTITION FIELD subdept",
+ tableName); // This spec is matching with the input data which is partitioned just by "id"
Review Comment:
// This spec now matches the partitioning of the parquet table
--
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: Add option to provide partition spec in ADD_FILES procedure [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on PR #12327:
URL: https://github.com/apache/iceberg/pull/12327#issuecomment-2669320344
As I noted on the issue, I think we can do this parameter-less if we try to find a compatible spec for whatever we are attempting to add. I know we probably need more extensive testing there as well but I think that makes more sense that manually specifying
--
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: Infer partition spec in ADD_FILES procedure for FileTables than taking latest table spec [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12327:
URL: https://github.com/apache/iceberg/pull/12327#discussion_r1968556347
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAddFilesProcedure.java:
##########
@@ -635,6 +635,67 @@ public void addFilteredPartitionsToPartitionedWithNullValueFilteringOnDept() {
sql("SELECT id, name, dept, subdept FROM %s ORDER BY id", tableName));
}
+ @TestTemplate
+ public void addFileTableOldSpecDataAfterPartitionSpecEvolved() {
+ createPartitionedFileTable("parquet");
+ createIcebergTable(
+ "id Integer, name String, dept String, subdept String",
+ "PARTITIONED BY (id, dept, subdept)");
+ sql("ALTER TABLE %s DROP PARTITION FIELD dept", tableName);
+ sql(
+ "ALTER TABLE %s DROP PARTITION FIELD subdept",
+ tableName); // This spec is matching with the input data which is partitioned just by "id"
+ sql("ALTER TABLE %s ADD PARTITION FIELD subdept", tableName);
+
+ if (formatVersion == 1) {
+ assertThatThrownBy(
+ () ->
+ scalarSql(
+ "CALL %s.system.add_files('%s', '`parquet`.`%s`')",
+ catalogName, tableName, fileTableDir.getAbsolutePath()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(
+ String.format(
+ "Cannot add data files to target table %s because that table is partitioned and contains non-identity partition transforms which will not be compatible.",
+ tableName));
+ return;
+ }
+
+ List<Object[]> result =
+ sql(
+ "CALL %s.system.add_files(table => '%s', source_table => '`parquet`.`%s`')",
+ catalogName, tableName, fileTableDir.getAbsolutePath());
+
+ assertOutput(result, 8L, 4L);
+ assertEquals(
+ "Iceberg table contains correct data",
+ sql("SELECT id, name, dept, subdept FROM %s ORDER BY id", sourceTableName),
+ sql("SELECT id, name, dept, subdept FROM %s ORDER BY id", tableName));
Review Comment:
Please also add in a check where we check the metadata tables and show that the correct spec id has been set for the newly added files. You can get the specId programmatically after you make it above with table.specId() and then assert that it is set for the new files 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: Infer partition spec in ADD_FILES procedure for FileTables than taking latest table spec [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12327:
URL: https://github.com/apache/iceberg/pull/12327#discussion_r1968558602
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/AddFilesProcedure.java:
##########
@@ -265,55 +271,16 @@ public String description() {
return "AddFiles";
}
- private void validatePartitionSpec(Table table, Map<String, String> partitionFilter) {
- List<PartitionField> partitionFields = table.spec().fields();
- Set<String> partitionNames =
- table.spec().fields().stream().map(PartitionField::name).collect(Collectors.toSet());
-
- boolean tablePartitioned = !partitionFields.isEmpty();
- boolean partitionSpecPassed = !partitionFilter.isEmpty();
-
+ private void nonIdentityPartitionCheck(Table table) {
// Check for any non-identity partition columns
List<PartitionField> nonIdentityFields =
- partitionFields.stream()
+ table.spec().fields().stream()
Review Comment:
This would be wrong then, because we are adding files with a different spec id. It's ok to have a current spec with a different partitioning with nonIdentityFields
--
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: Infer partition spec in ADD_FILES procedure for FileTables than taking latest table spec [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12327:
URL: https://github.com/apache/iceberg/pull/12327#discussion_r1968554090
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/AddFilesProcedure.java:
##########
@@ -196,11 +194,23 @@ private void importFileTable(
String format,
Map<String, String> partitionFilter,
boolean checkDuplicateFiles,
- PartitionSpec spec,
int parallelism) {
+
+ org.apache.spark.sql.execution.datasources.PartitionSpec inferredSpec =
+ getInferredSpec(spark(), tableLocation);
+
+ List<String> sparkPartNames =
+ JavaConverters.seqAsJavaList(inferredSpec.partitionColumns()).stream()
+ .map(StructField::name)
+ .map(name -> name.toLowerCase(Locale.ROOT))
Review Comment:
don't we lower case these in findCompatibleSpec?
--
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: Infer partition spec in ADD_FILES procedure for FileTables than taking latest table spec [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12327:
URL: https://github.com/apache/iceberg/pull/12327#discussion_r1970303712
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAddFilesProcedure.java:
##########
@@ -635,6 +635,67 @@ public void addFilteredPartitionsToPartitionedWithNullValueFilteringOnDept() {
sql("SELECT id, name, dept, subdept FROM %s ORDER BY id", tableName));
}
+ @TestTemplate
+ public void addFileTableOldSpecDataAfterPartitionSpecEvolved() {
+ createPartitionedFileTable("parquet");
+ createIcebergTable(
+ "id Integer, name String, dept String, subdept String",
+ "PARTITIONED BY (id, dept, subdept)");
+ sql("ALTER TABLE %s DROP PARTITION FIELD dept", tableName);
+ sql(
+ "ALTER TABLE %s DROP PARTITION FIELD subdept",
+ tableName); // This spec is matching with the input data which is partitioned just by "id"
+ sql("ALTER TABLE %s ADD PARTITION FIELD subdept", tableName);
+
+ if (formatVersion == 1) {
+ assertThatThrownBy(
Review Comment:
Fair enough
--
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: Add option to provide partition spec in ADD_FILES procedure [iceberg]
Posted by "bharos (via GitHub)" <gi...@apache.org>.
bharos commented on code in PR #12327:
URL: https://github.com/apache/iceberg/pull/12327#discussion_r1961148678
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAddFilesProcedure.java:
##########
@@ -607,6 +607,40 @@ public void addFilteredPartitionsToPartitionedWithNullValueFilteringOnDept() {
sql("SELECT id, name, dept, subdept FROM %s ORDER BY id", tableName));
}
+ @TestTemplate
+ public void addFileTableInvalidPartitionSpecVersion() {
+ createPartitionedFileTable("parquet");
+ createIcebergTable(
+ "id Integer, name String, dept String, subdept String", "PARTITIONED BY (id, dept)");
+
+ assertThatThrownBy(
+ () ->
+ sql(
+ "CALL %s.system.add_files('%s', '`parquet`.`%s`', map('id', 1), true, 1, 1)",
+ catalogName, tableName, fileTableDir.getAbsolutePath()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid partition spec version: 1");
+ }
+
+ @TestTemplate
+ public void addFileTableWithPartitionSpecVersion() {
+ createPartitionedFileTable("parquet");
+ createIcebergTable(
+ "id Integer, name String, dept String, subdept String", "PARTITIONED BY (id)");
+ sql("ALTER TABLE %s ADD PARTITION FIELD dept", tableName);
+
+ List<Object[]> result =
+ sql(
+ "CALL %s.system.add_files(table => '%s', source_table => '`parquet`.`%s`', partition_spec_version => 0)",
+ catalogName, tableName, fileTableDir.getAbsolutePath());
Review Comment:
This test fails if we don't add `partition_spec_version => 0` which is why adding this feature
--
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: Infer partition spec in ADD_FILES procedure rather than taking latest table spec [iceberg]
Posted by "bharos (via GitHub)" <gi...@apache.org>.
bharos commented on code in PR #12327:
URL: https://github.com/apache/iceberg/pull/12327#discussion_r1962601041
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/AddFilesProcedure.java:
##########
@@ -265,55 +271,16 @@ public String description() {
return "AddFiles";
}
- private void validatePartitionSpec(Table table, Map<String, String> partitionFilter) {
- List<PartitionField> partitionFields = table.spec().fields();
- Set<String> partitionNames =
- table.spec().fields().stream().map(PartitionField::name).collect(Collectors.toSet());
-
- boolean tablePartitioned = !partitionFields.isEmpty();
- boolean partitionSpecPassed = !partitionFilter.isEmpty();
-
+ private void nonIdentityPartitionCheck(Table table) {
// Check for any non-identity partition columns
List<PartitionField> nonIdentityFields =
- partitionFields.stream()
+ table.spec().fields().stream()
Review Comment:
Using table.spec() here to determine nonIdentityFields as we want to see if the table has any non-identity partitions until 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: Infer partition spec in ADD_FILES procedure rather than taking latest table spec [iceberg]
Posted by "bharos (via GitHub)" <gi...@apache.org>.
bharos commented on code in PR #12327:
URL: https://github.com/apache/iceberg/pull/12327#discussion_r1962465950
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAddFilesProcedure.java:
##########
@@ -607,6 +607,40 @@ public void addFilteredPartitionsToPartitionedWithNullValueFilteringOnDept() {
sql("SELECT id, name, dept, subdept FROM %s ORDER BY id", tableName));
}
+ @TestTemplate
+ public void addFileTableInvalidPartitionSpecVersion() {
+ createPartitionedFileTable("parquet");
+ createIcebergTable(
+ "id Integer, name String, dept String, subdept String", "PARTITIONED BY (id, dept)");
+
+ assertThatThrownBy(
+ () ->
+ sql(
+ "CALL %s.system.add_files('%s', '`parquet`.`%s`', map('id', 1), true, 1, 1)",
+ catalogName, tableName, fileTableDir.getAbsolutePath()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid partition spec version: 1");
+ }
+
+ @TestTemplate
+ public void addFileTableWithPartitionSpecVersion() {
+ createPartitionedFileTable("parquet");
+ createIcebergTable(
+ "id Integer, name String, dept String, subdept String", "PARTITIONED BY (id)");
+ sql("ALTER TABLE %s ADD PARTITION FIELD dept", tableName);
+
+ List<Object[]> result =
+ sql(
+ "CALL %s.system.add_files(table => '%s', source_table => '`parquet`.`%s`', partition_spec_version => 0)",
+ catalogName, tableName, fileTableDir.getAbsolutePath());
Review Comment:
N/A with latest 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: Infer partition spec in ADD_FILES procedure for FileTables than taking latest table spec [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12327:
URL: https://github.com/apache/iceberg/pull/12327#discussion_r1970309065
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkTableUtil.java:
##########
@@ -1124,7 +1115,67 @@ private static PartitionSpec findCompatibleSpec(
throw new IllegalArgumentException(
String.format(
"Cannot find a partition spec in Iceberg table %s that matches the partition"
- + " columns (%s) in Spark table %s",
- icebergTable, sparkPartNames, sparkTable));
+ + " columns (%s) in input table",
+ icebergTable, partitionNames));
+ }
+
+ /**
+ * Returns the first partition spec in an IcebergTable that shares the same names and ordering as
+ * the partition columns in a given Spark Table. Throws an error if not found
+ */
+ private static PartitionSpec findCompatibleSpec(
+ Table icebergTable, SparkSession spark, String sparkTable) throws AnalysisException {
+ List<String> parts = Lists.newArrayList(Splitter.on('.').limit(2).split(sparkTable));
+ String db = parts.size() == 1 ? "default" : parts.get(0);
+ String table = parts.get(parts.size() == 1 ? 0 : 1);
+
+ List<String> sparkPartNames =
+ spark.catalog().listColumns(db, table).collectAsList().stream()
+ .filter(org.apache.spark.sql.catalog.Column::isPartition)
+ .map(org.apache.spark.sql.catalog.Column::name)
+ .map(name -> name.toLowerCase(Locale.ROOT))
+ .collect(Collectors.toList());
+ return findCompatibleSpec(sparkPartNames, icebergTable);
+ }
+
+ public static void validatePartitionFilter(
Review Comment:
I miss Scala, public is fine then.
--
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: Infer partition spec in ADD_FILES procedure for FileTables than taking latest table spec [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12327:
URL: https://github.com/apache/iceberg/pull/12327#discussion_r1968551184
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkTableUtil.java:
##########
@@ -1124,7 +1115,67 @@ private static PartitionSpec findCompatibleSpec(
throw new IllegalArgumentException(
String.format(
"Cannot find a partition spec in Iceberg table %s that matches the partition"
- + " columns (%s) in Spark table %s",
- icebergTable, sparkPartNames, sparkTable));
+ + " columns (%s) in input table",
+ icebergTable, partitionNames));
+ }
+
+ /**
+ * Returns the first partition spec in an IcebergTable that shares the same names and ordering as
+ * the partition columns in a given Spark Table. Throws an error if not found
+ */
+ private static PartitionSpec findCompatibleSpec(
+ Table icebergTable, SparkSession spark, String sparkTable) throws AnalysisException {
+ List<String> parts = Lists.newArrayList(Splitter.on('.').limit(2).split(sparkTable));
+ String db = parts.size() == 1 ? "default" : parts.get(0);
+ String table = parts.get(parts.size() == 1 ? 0 : 1);
+
+ List<String> sparkPartNames =
+ spark.catalog().listColumns(db, table).collectAsList().stream()
+ .filter(org.apache.spark.sql.catalog.Column::isPartition)
+ .map(org.apache.spark.sql.catalog.Column::name)
+ .map(name -> name.toLowerCase(Locale.ROOT))
Review Comment:
We already lower case these in the polymorphic version so we probably don't need to call it 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: Infer partition spec in ADD_FILES procedure for FileTables than taking latest table spec [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12327:
URL: https://github.com/apache/iceberg/pull/12327#discussion_r1968561130
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/Spark3Util.java:
##########
@@ -858,6 +858,23 @@ public static String quotedFullIdentifier(String catalogName, Identifier identif
.quoted();
}
+ public static org.apache.spark.sql.execution.datasources.PartitionSpec getInferredSpec(
+ SparkSession spark, Path rootPath) {
+ FileStatusCache fileStatusCache = FileStatusCache.getOrCreate(spark);
+ InMemoryFileIndex fileIndex =
+ new InMemoryFileIndex(
Review Comment:
... I think this is ok .. I worry a little bit about making caches that we never clear but I don't think we have a safe way of invalidating this cache either.
--
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: Infer partition spec in ADD_FILES procedure for FileTables than taking latest table spec [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on PR #12327:
URL: https://github.com/apache/iceberg/pull/12327#issuecomment-2682938175
* What went wrong:
Execution failed for task ':iceberg-spark:iceberg-spark-3.5_2.12:checkstyleMain'.
> A failure occurred while executing org.gradle.api.plugins.quality.internal.CheckstyleAction
> Checkstyle rule violations were found. See the report at: file:///home/runner/work/iceberg/iceberg/spark/v3.5/spark/build/reports/checkstyle/main.html
Checkstyle files with violations: 1
Checkstyle violations by severity: [error:3]
Make sure to run spotlessApply to fix the style issues
--
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: Infer partition spec in ADD_FILES procedure rather than taking latest table spec [iceberg]
Posted by "bharos (via GitHub)" <gi...@apache.org>.
bharos commented on code in PR #12327:
URL: https://github.com/apache/iceberg/pull/12327#discussion_r1962469451
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/AddFilesProcedure.java:
##########
@@ -196,11 +193,36 @@ private void importFileTable(
String format,
Map<String, String> partitionFilter,
boolean checkDuplicateFiles,
- PartitionSpec spec,
int parallelism) {
+
+ org.apache.spark.sql.execution.datasources.PartitionSpec inferredSpec =
+ getInferredSpec(spark(), tableLocation);
+
+ List<String> sparkPartNames =
+ JavaConverters.seqAsJavaList(inferredSpec.partitionColumns()).stream()
+ .map(StructField::name)
+ .map(name -> name.toLowerCase(Locale.ROOT))
+ .collect(Collectors.toList());
+ PartitionSpec matchingSpec = null;
+ for (PartitionSpec icebergSpec : table.specs().values()) {
+ List<String> icebergPartNames =
+ icebergSpec.fields().stream()
+ .map(PartitionField::name)
+ .map(name -> name.toLowerCase(Locale.ROOT))
+ .collect(Collectors.toList());
+ if (icebergPartNames.equals(sparkPartNames)) {
+ matchingSpec = icebergSpec;
+ break;
+ }
+ }
+ if (matchingSpec == null) {
+ // If the inferred spec does not match any existing spec, use the Iceberg table's latest spec
+ matchingSpec = table.spec();
Review Comment:
Taking the table spec if we don't find any matches.
Should we throw an error otherwise?
--
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: Infer partition spec in ADD_FILES procedure rather than taking latest table spec [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12327:
URL: https://github.com/apache/iceberg/pull/12327#discussion_r1962474190
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/AddFilesProcedure.java:
##########
@@ -196,11 +193,36 @@ private void importFileTable(
String format,
Map<String, String> partitionFilter,
boolean checkDuplicateFiles,
- PartitionSpec spec,
int parallelism) {
+
+ org.apache.spark.sql.execution.datasources.PartitionSpec inferredSpec =
+ getInferredSpec(spark(), tableLocation);
+
+ List<String> sparkPartNames =
+ JavaConverters.seqAsJavaList(inferredSpec.partitionColumns()).stream()
+ .map(StructField::name)
+ .map(name -> name.toLowerCase(Locale.ROOT))
+ .collect(Collectors.toList());
+ PartitionSpec matchingSpec = null;
+ for (PartitionSpec icebergSpec : table.specs().values()) {
+ List<String> icebergPartNames =
+ icebergSpec.fields().stream()
+ .map(PartitionField::name)
+ .map(name -> name.toLowerCase(Locale.ROOT))
+ .collect(Collectors.toList());
+ if (icebergPartNames.equals(sparkPartNames)) {
+ matchingSpec = icebergSpec;
+ break;
+ }
+ }
+ if (matchingSpec == null) {
+ // If the inferred spec does not match any existing spec, use the Iceberg table's latest spec
+ matchingSpec = table.spec();
Review Comment:
I think this is an error 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: Infer partition spec in ADD_FILES procedure rather than taking latest table spec [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12327:
URL: https://github.com/apache/iceberg/pull/12327#discussion_r1962477700
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/AddFilesProcedure.java:
##########
@@ -196,11 +193,36 @@ private void importFileTable(
String format,
Map<String, String> partitionFilter,
boolean checkDuplicateFiles,
- PartitionSpec spec,
int parallelism) {
+
+ org.apache.spark.sql.execution.datasources.PartitionSpec inferredSpec =
+ getInferredSpec(spark(), tableLocation);
+
+ List<String> sparkPartNames =
Review Comment:
I would use the SparkTableUtil method we just merged in the other pr. Feel free to modify it as needed
--
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: Infer partition spec in ADD_FILES procedure for FileTables than taking latest table spec [iceberg]
Posted by "bharos (via GitHub)" <gi...@apache.org>.
bharos commented on code in PR #12327:
URL: https://github.com/apache/iceberg/pull/12327#discussion_r1968579723
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAddFilesProcedure.java:
##########
@@ -635,6 +635,67 @@ public void addFilteredPartitionsToPartitionedWithNullValueFilteringOnDept() {
sql("SELECT id, name, dept, subdept FROM %s ORDER BY id", tableName));
}
+ @TestTemplate
+ public void addFileTableOldSpecDataAfterPartitionSpecEvolved() {
+ createPartitionedFileTable("parquet");
+ createIcebergTable(
+ "id Integer, name String, dept String, subdept String",
+ "PARTITIONED BY (id, dept, subdept)");
+ sql("ALTER TABLE %s DROP PARTITION FIELD dept", tableName);
+ sql(
+ "ALTER TABLE %s DROP PARTITION FIELD subdept",
+ tableName); // This spec is matching with the input data which is partitioned just by "id"
+ sql("ALTER TABLE %s ADD PARTITION FIELD subdept", tableName);
+
+ if (formatVersion == 1) {
+ assertThatThrownBy(
Review Comment:
I think keeping this will help understand that failing in V1 is expected, at least for somebody not familiar with the void transforms in v1
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/AddFilesProcedure.java:
##########
@@ -265,55 +271,16 @@ public String description() {
return "AddFiles";
}
- private void validatePartitionSpec(Table table, Map<String, String> partitionFilter) {
- List<PartitionField> partitionFields = table.spec().fields();
- Set<String> partitionNames =
- table.spec().fields().stream().map(PartitionField::name).collect(Collectors.toSet());
-
- boolean tablePartitioned = !partitionFields.isEmpty();
- boolean partitionSpecPassed = !partitionFilter.isEmpty();
-
+ private void nonIdentityPartitionCheck(Table table) {
// Check for any non-identity partition columns
List<PartitionField> nonIdentityFields =
- partitionFields.stream()
+ table.spec().fields().stream()
Review Comment:
Removing this method as per above discussion
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkTableUtil.java:
##########
@@ -1124,7 +1115,67 @@ private static PartitionSpec findCompatibleSpec(
throw new IllegalArgumentException(
String.format(
"Cannot find a partition spec in Iceberg table %s that matches the partition"
- + " columns (%s) in Spark table %s",
- icebergTable, sparkPartNames, sparkTable));
+ + " columns (%s) in input table",
+ icebergTable, partitionNames));
+ }
+
+ /**
+ * Returns the first partition spec in an IcebergTable that shares the same names and ordering as
+ * the partition columns in a given Spark Table. Throws an error if not found
+ */
+ private static PartitionSpec findCompatibleSpec(
+ Table icebergTable, SparkSession spark, String sparkTable) throws AnalysisException {
+ List<String> parts = Lists.newArrayList(Splitter.on('.').limit(2).split(sparkTable));
+ String db = parts.size() == 1 ? "default" : parts.get(0);
+ String table = parts.get(parts.size() == 1 ? 0 : 1);
+
+ List<String> sparkPartNames =
+ spark.catalog().listColumns(db, table).collectAsList().stream()
+ .filter(org.apache.spark.sql.catalog.Column::isPartition)
+ .map(org.apache.spark.sql.catalog.Column::name)
+ .map(name -> name.toLowerCase(Locale.ROOT))
Review Comment:
done
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAddFilesProcedure.java:
##########
@@ -635,6 +635,67 @@ public void addFilteredPartitionsToPartitionedWithNullValueFilteringOnDept() {
sql("SELECT id, name, dept, subdept FROM %s ORDER BY id", tableName));
}
+ @TestTemplate
+ public void addFileTableOldSpecDataAfterPartitionSpecEvolved() {
+ createPartitionedFileTable("parquet");
+ createIcebergTable(
+ "id Integer, name String, dept String, subdept String",
+ "PARTITIONED BY (id, dept, subdept)");
+ sql("ALTER TABLE %s DROP PARTITION FIELD dept", tableName);
+ sql(
+ "ALTER TABLE %s DROP PARTITION FIELD subdept",
+ tableName); // This spec is matching with the input data which is partitioned just by "id"
+ sql("ALTER TABLE %s ADD PARTITION FIELD subdept", tableName);
+
+ if (formatVersion == 1) {
+ assertThatThrownBy(
+ () ->
+ scalarSql(
+ "CALL %s.system.add_files('%s', '`parquet`.`%s`')",
+ catalogName, tableName, fileTableDir.getAbsolutePath()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(
+ String.format(
+ "Cannot add data files to target table %s because that table is partitioned and contains non-identity partition transforms which will not be compatible.",
+ tableName));
+ return;
+ }
+
+ List<Object[]> result =
+ sql(
+ "CALL %s.system.add_files(table => '%s', source_table => '`parquet`.`%s`')",
+ catalogName, tableName, fileTableDir.getAbsolutePath());
+
+ assertOutput(result, 8L, 4L);
+ assertEquals(
+ "Iceberg table contains correct data",
+ sql("SELECT id, name, dept, subdept FROM %s ORDER BY id", sourceTableName),
+ sql("SELECT id, name, dept, subdept FROM %s ORDER BY id", tableName));
Review Comment:
Added
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/AddFilesProcedure.java:
##########
@@ -265,55 +271,16 @@ public String description() {
return "AddFiles";
}
- private void validatePartitionSpec(Table table, Map<String, String> partitionFilter) {
- List<PartitionField> partitionFields = table.spec().fields();
- Set<String> partitionNames =
- table.spec().fields().stream().map(PartitionField::name).collect(Collectors.toSet());
-
- boolean tablePartitioned = !partitionFields.isEmpty();
- boolean partitionSpecPassed = !partitionFilter.isEmpty();
-
+ private void nonIdentityPartitionCheck(Table table) {
Review Comment:
Makes sense. Since we check if (allIdentity) in findCompatibleSpec, we can remove this one
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkTableUtil.java:
##########
@@ -1124,7 +1115,67 @@ private static PartitionSpec findCompatibleSpec(
throw new IllegalArgumentException(
String.format(
"Cannot find a partition spec in Iceberg table %s that matches the partition"
- + " columns (%s) in Spark table %s",
- icebergTable, sparkPartNames, sparkTable));
+ + " columns (%s) in input table",
+ icebergTable, partitionNames));
+ }
+
+ /**
+ * Returns the first partition spec in an IcebergTable that shares the same names and ordering as
+ * the partition columns in a given Spark Table. Throws an error if not found
+ */
+ private static PartitionSpec findCompatibleSpec(
+ Table icebergTable, SparkSession spark, String sparkTable) throws AnalysisException {
+ List<String> parts = Lists.newArrayList(Splitter.on('.').limit(2).split(sparkTable));
+ String db = parts.size() == 1 ? "default" : parts.get(0);
+ String table = parts.get(parts.size() == 1 ? 0 : 1);
+
+ List<String> sparkPartNames =
+ spark.catalog().listColumns(db, table).collectAsList().stream()
+ .filter(org.apache.spark.sql.catalog.Column::isPartition)
+ .map(org.apache.spark.sql.catalog.Column::name)
+ .map(name -> name.toLowerCase(Locale.ROOT))
+ .collect(Collectors.toList());
+ return findCompatibleSpec(sparkPartNames, icebergTable);
+ }
+
+ public static void validatePartitionFilter(
Review Comment:
AddFilesProcedure is in package org.apache.iceberg.spark.procedures;
This one is in package org.apache.iceberg.spark;
Is there any common package where this can be moved to ? Otherwise will keep it public here for both findCompatibleSpec and validatePartitionFilter
--
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: Infer partition spec in ADD_FILES procedure rather than taking latest table spec [iceberg]
Posted by "bharos (via GitHub)" <gi...@apache.org>.
bharos commented on code in PR #12327:
URL: https://github.com/apache/iceberg/pull/12327#discussion_r1962595085
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/AddFilesProcedure.java:
##########
@@ -196,11 +193,36 @@ private void importFileTable(
String format,
Map<String, String> partitionFilter,
boolean checkDuplicateFiles,
- PartitionSpec spec,
int parallelism) {
+
+ org.apache.spark.sql.execution.datasources.PartitionSpec inferredSpec =
+ getInferredSpec(spark(), tableLocation);
+
+ List<String> sparkPartNames =
+ JavaConverters.seqAsJavaList(inferredSpec.partitionColumns()).stream()
+ .map(StructField::name)
+ .map(name -> name.toLowerCase(Locale.ROOT))
+ .collect(Collectors.toList());
+ PartitionSpec matchingSpec = null;
+ for (PartitionSpec icebergSpec : table.specs().values()) {
+ List<String> icebergPartNames =
+ icebergSpec.fields().stream()
+ .map(PartitionField::name)
+ .map(name -> name.toLowerCase(Locale.ROOT))
+ .collect(Collectors.toList());
+ if (icebergPartNames.equals(sparkPartNames)) {
+ matchingSpec = icebergSpec;
+ break;
+ }
+ }
+ if (matchingSpec == null) {
+ // If the inferred spec does not match any existing spec, use the Iceberg table's latest spec
+ matchingSpec = table.spec();
Review Comment:
Using the findCompatibleSpec method from SparkTableUtil. so now this will throw an error
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/AddFilesProcedure.java:
##########
@@ -265,13 +283,13 @@ public String description() {
return "AddFiles";
}
- private void validatePartitionSpec(Table table, Map<String, String> partitionFilter) {
+ private void validatePartitionFilter(Table table, Map<String, String> partitionFilter) {
List<PartitionField> partitionFields = table.spec().fields();
Set<String> partitionNames =
table.spec().fields().stream().map(PartitionField::name).collect(Collectors.toSet());
Review Comment:
Fixed this to take partition spec and not always take the table latest spec.
Although, I had to make a refactor here, I moved the identity partition check from this function to another function so that I can check it at the beginning itself, because we can just exit early in that case.
Then, I am doing the `validatePartitionFilter` later, after we find a compatible spec (because now we have to pass it to this function)
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/AddFilesProcedure.java:
##########
@@ -196,11 +193,36 @@ private void importFileTable(
String format,
Map<String, String> partitionFilter,
boolean checkDuplicateFiles,
- PartitionSpec spec,
int parallelism) {
+
+ org.apache.spark.sql.execution.datasources.PartitionSpec inferredSpec =
+ getInferredSpec(spark(), tableLocation);
+
+ List<String> sparkPartNames =
Review Comment:
Thanks, I refactored the part that's commonly needed to a public method. PTAL
--
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: Infer partition spec in ADD_FILES procedure for FileTables than taking latest table spec [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer merged PR #12327:
URL: https://github.com/apache/iceberg/pull/12327
--
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