You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by bl...@apache.org on 2021/12/21 20:26:56 UTC
[iceberg] branch master updated: Spark 3.1: Fix NPE in add_files with null partition values (#3785)
This is an automated email from the ASF dual-hosted git repository.
blue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/master by this push:
new 2aa3fc7 Spark 3.1: Fix NPE in add_files with null partition values (#3785)
2aa3fc7 is described below
commit 2aa3fc7690e0b5ab70e0ff7cffe404cc929796c5
Author: Kyle Bendickson <kj...@gmail.com>
AuthorDate: Tue Dec 21 12:26:46 2021 -0800
Spark 3.1: Fix NPE in add_files with null partition values (#3785)
---
.../spark/extensions/TestAddFilesProcedure.java | 70 ++++++++++++++++++++--
.../java/org/apache/iceberg/spark/Spark3Util.java | 2 +-
2 files changed, 67 insertions(+), 5 deletions(-)
diff --git a/spark/v3.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAddFilesProcedure.java b/spark/v3.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAddFilesProcedure.java
index 7162294..215f15a 100644
--- a/spark/v3.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAddFilesProcedure.java
+++ b/spark/v3.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAddFilesProcedure.java
@@ -412,6 +412,46 @@ public class TestAddFilesProcedure extends SparkExtensionsTestBase {
}
@Test
+ public void addFilteredPartitionsToPartitionedWithNullValueFilteringOnId() {
+ createCompositePartitionedTableWithNullValueInPartitionColumn("parquet");
+
+ String createIceberg =
+ "CREATE TABLE %s (id Integer, name String, dept String, subdept String) USING iceberg " +
+ "PARTITIONED BY (id, dept)";
+
+ sql(createIceberg, tableName);
+
+ Object result = scalarSql("CALL %s.system.add_files('%s', '`parquet`.`%s`', map('id', 1))",
+ catalogName, tableName, fileTableDir.getAbsolutePath());
+
+ Assert.assertEquals(2L, result);
+
+ assertEquals("Iceberg table contains correct data",
+ sql("SELECT id, name, dept, subdept FROM %s WHERE id = 1 ORDER BY id", sourceTableName),
+ sql("SELECT id, name, dept, subdept FROM %s ORDER BY id", tableName));
+ }
+
+ @Test
+ public void addFilteredPartitionsToPartitionedWithNullValueFilteringOnDept() {
+ createCompositePartitionedTableWithNullValueInPartitionColumn("parquet");
+
+ String createIceberg =
+ "CREATE TABLE %s (id Integer, name String, dept String, subdept String) USING iceberg " +
+ "PARTITIONED BY (id, dept)";
+
+ sql(createIceberg, tableName);
+
+ Object result = scalarSql("CALL %s.system.add_files('%s', '`parquet`.`%s`', map('dept', 'hr'))",
+ catalogName, tableName, fileTableDir.getAbsolutePath());
+
+ Assert.assertEquals(6L, result);
+
+ assertEquals("Iceberg table contains correct data",
+ sql("SELECT id, name, dept, subdept FROM %s WHERE dept = 'hr' ORDER BY id", sourceTableName),
+ sql("SELECT id, name, dept, subdept FROM %s ORDER BY id", tableName));
+ }
+
+ @Test
public void addWeirdCaseHiveTable() {
createWeirdCaseTable();
@@ -748,10 +788,10 @@ public class TestAddFilesProcedure extends SparkExtensionsTestBase {
private static final List<Object[]> emptyQueryResult = Lists.newArrayList();
private static final StructField[] struct = {
- new StructField("id", DataTypes.IntegerType, false, Metadata.empty()),
- new StructField("name", DataTypes.StringType, false, Metadata.empty()),
- new StructField("dept", DataTypes.StringType, false, Metadata.empty()),
- new StructField("subdept", DataTypes.StringType, false, Metadata.empty())
+ new StructField("id", DataTypes.IntegerType, true, Metadata.empty()),
+ new StructField("name", DataTypes.StringType, true, Metadata.empty()),
+ new StructField("dept", DataTypes.StringType, true, Metadata.empty()),
+ new StructField("subdept", DataTypes.StringType, true, Metadata.empty())
};
private static final Dataset<Row> unpartitionedDF =
@@ -763,12 +803,21 @@ public class TestAddFilesProcedure extends SparkExtensionsTestBase {
RowFactory.create(4, "Will Doe", "facilities", "all")),
new StructType(struct)).repartition(1);
+ private static final Dataset<Row> singleNullRecordDF =
+ spark.createDataFrame(
+ ImmutableList.of(
+ RowFactory.create(null, null, null, null)),
+ new StructType(struct)).repartition(1);
+
private static final Dataset<Row> partitionedDF =
unpartitionedDF.select("name", "dept", "subdept", "id");
private static final Dataset<Row> compositePartitionedDF =
unpartitionedDF.select("name", "subdept", "id", "dept");
+ private static final Dataset<Row> compositePartitionedNullRecordDF =
+ singleNullRecordDF.select("name", "subdept", "id", "dept");
+
private static final Dataset<Row> weirdColumnNamesDF =
unpartitionedDF.select(
unpartitionedDF.col("id"),
@@ -806,6 +855,19 @@ public class TestAddFilesProcedure extends SparkExtensionsTestBase {
compositePartitionedDF.write().insertInto(sourceTableName);
}
+ private void createCompositePartitionedTableWithNullValueInPartitionColumn(String format) {
+ String createParquet = "CREATE TABLE %s (id Integer, name String, dept String, subdept String) USING %s " +
+ "PARTITIONED BY (id, dept) LOCATION '%s'";
+ sql(createParquet, sourceTableName, format, fileTableDir.getAbsolutePath());
+
+ Dataset<Row> unionedDF = compositePartitionedDF.unionAll(compositePartitionedNullRecordDF)
+ .select("name", "subdept", "id", "dept")
+ .repartition(1);
+
+ unionedDF.write().insertInto(sourceTableName);
+ unionedDF.write().insertInto(sourceTableName);
+ }
+
private void createWeirdCaseTable() {
String createParquet =
"CREATE TABLE %s (id Integer, subdept String, dept String) " +
diff --git a/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/Spark3Util.java b/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/Spark3Util.java
index 215966c..7ba3bbb 100644
--- a/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/Spark3Util.java
+++ b/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/Spark3Util.java
@@ -849,7 +849,7 @@ public class Spark3Util {
int fieldIndex = schema.fieldIndex(field.name());
Object catalystValue = partition.values().get(fieldIndex, field.dataType());
Object value = CatalystTypeConverters.convertToScala(catalystValue, field.dataType());
- values.put(field.name(), value.toString());
+ values.put(field.name(), String.valueOf(value));
});
return new SparkPartition(values, partition.path().toString(), format);
}).collect(Collectors.toList());