Skip to content

Push down partition filter to Spark when Importing File Based Tables #3745

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rdblue merged 11 commits into from
Jan 12, 2022
Merged

Push down partition filter to Spark when Importing File Based Tables #3745

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
3ef0170
Push down partition filter to Spark when Importing File Based Tables
huaxingao Dec 14, 2021
46602b8
use case for condition + Lists.newArrayList()
huaxingao Dec 15, 2021
f72721e
fix indentation
huaxingao Dec 15, 2021
79828e2
remove unused import
huaxingao Dec 15, 2021
51ee198
use SparkExpressionConverter.collectResolvedSparkExpression to conver…
huaxingao Dec 23, 2021
19cb5c7
address comments
huaxingao Dec 23, 2021
7d95086
fix precondition message
huaxingao Jan 7, 2022
ffc5c6e
address comments
huaxingao Jan 12, 2022
3b4ba27
ReTrigger Build
huaxingao Jan 12, 2022
f9b2f40
Trigger Build
huaxingao Jan 12, 2022
f8822e4
address comments
huaxingao Jan 12, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 42 additions & 3 deletions spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/Spark3Util.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.iceberg.MetadataTableType;
import org.apache.iceberg.MetadataTableUtils;
Expand All @@ -48,6 +49,7 @@
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.spark.SparkTableUtil.SparkPartition;
import org.apache.iceberg.spark.source.SparkTable;
Expand All @@ -57,6 +59,7 @@
import org.apache.iceberg.types.TypeUtil;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.util.Pair;
import org.apache.spark.sql.AnalysisException;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
Expand All @@ -77,6 +80,8 @@
import org.apache.spark.sql.connector.expressions.Transform;
import org.apache.spark.sql.execution.datasources.FileStatusCache;
import org.apache.spark.sql.execution.datasources.InMemoryFileIndex;
import org.apache.spark.sql.execution.datasources.PartitionDirectory;
import org.apache.spark.sql.execution.datasources.SparkExpressionConverter;
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation;
import org.apache.spark.sql.types.IntegerType;
import org.apache.spark.sql.types.LongType;
Expand Down Expand Up @@ -745,9 +750,11 @@ public static TableIdentifier identifierToTableIdentifier(Identifier identifier)
* @param spark a Spark session
* @param rootPath a table identifier
* @param format format of the file
* @param partitionFilter partitionFilter of the file
* @return all table's partitions
*/
public static List<SparkPartition> getPartitions(SparkSession spark, Path rootPath, String format) {
public static List<SparkPartition> getPartitions(SparkSession spark, String tableName, Path rootPath, String format,
Comment thread
rdblue marked this conversation as resolved.
Outdated
Map<String, String> partitionFilter) {
FileStatusCache fileStatusCache = FileStatusCache.getOrCreate(spark);
Map<String, String> emptyMap = Collections.emptyMap();

Expand All @@ -768,9 +775,22 @@ public static List<SparkPartition> getPartitions(SparkSession spark, Path rootPa

org.apache.spark.sql.execution.datasources.PartitionSpec spec = fileIndex.partitionSpec();
StructType schema = spec.partitionColumns();
if (schema.isEmpty()) {
return Lists.newArrayList();
}

List<org.apache.spark.sql.catalyst.expressions.Expression> filterExpressions =
getPartitionFilterExpressions(spark, tableName, partitionFilter);

List<org.apache.spark.sql.catalyst.expressions.Expression> dataFilters = Lists.newArrayList();
Seq<org.apache.spark.sql.catalyst.expressions.Expression> scalaPartitionFilters =
JavaConverters.asScalaBufferConverter(filterExpressions).asScala().toSeq();
Seq<org.apache.spark.sql.catalyst.expressions.Expression> scalaDataFilters =
JavaConverters.asScalaBufferConverter(dataFilters).asScala().toSeq();

@rdblue rdblue Jan 9, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is there an easier way to construct an empty sequence? Also, since this is always empty, can you put the dataFilters definition and this line next to one another? The line to create scalaPartitionFilters can be next to the line above that creates filterExpressions.

Seq<PartitionDirectory> filteredPartitions = fileIndex.listFiles(scalaPartitionFilters, scalaDataFilters);

return JavaConverters
.seqAsJavaListConverter(spec.partitions())
.seqAsJavaListConverter(filteredPartitions)
.asJava()
.stream()
.map(partition -> {
Expand All @@ -781,10 +801,29 @@ public static List<SparkPartition> getPartitions(SparkSession spark, Path rootPa
Object value = CatalystTypeConverters.convertToScala(catalystValue, field.dataType());
values.put(field.name(), String.valueOf(value));
});
return new SparkPartition(values, partition.path().toString(), format);
FileStatus fileStatus =
Comment thread
rdblue marked this conversation as resolved.
scala.collection.JavaConverters.seqAsJavaListConverter(partition.files()).asJava().get(0);

@rdblue rdblue Jan 9, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

scala.collection.JavaConverters is imported. Can you remove the fully-qualified name?


return new SparkPartition(values, fileStatus.getPath().getParent().toString(), format);
}).collect(Collectors.toList());
}

private static List<org.apache.spark.sql.catalyst.expressions.Expression> getPartitionFilterExpressions(

@rdblue rdblue Jan 9, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is it possible to move this to a separate util class to avoid the conflict with the connector Expression? Maybe SparkPartitionUtil or something?

SparkSession spark, String tableName, Map<String, String> partitionFilter) {
List<org.apache.spark.sql.catalyst.expressions.Expression> filterExpressions = Lists.newArrayList();
for (Map.Entry<String, String> entry : partitionFilter.entrySet()) {
String filter = entry.getKey() + " = '" + entry.getValue() + "'";

@ajantha-bhat ajantha-bhat Dec 23, 2021 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why do we have quotes ('') for the value?
Example: If id is an integer column, then we need id = 3 in the query instead of id ='3' ?

have we tested both string and non string partition columns ?

I was also thinking instead of map, can we expose the where clause in the call procedure (similar to rewrite_data files), so user can give filters other than equals also?

@ajantha-bhat ajantha-bhat Dec 23, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I was also thinking instead of map, can we expose the where clause in the call procedure (similar to rewrite_data files), so user can give filters other than equals also?

But it will become breaking API changes I guess. Let's see what others think on this also.

@huaxingao huaxingao Dec 23, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If the filter is on String column, e.g. dept column with value hr, we want the filter to be dept = 'hr'.

For non-String columns, such as id = 3, the filter with id = '3' still works ok because Spark will have Literal with String value to begin with, and then cast to Literal with Int value after it has the column type.

TestAddFilesProcedure has both String and int type partition columns so these two are tested. We probably should test other types e.g. Timestamp just to make sure.

try {
org.apache.spark.sql.catalyst.expressions.Expression expression =
SparkExpressionConverter.collectResolvedSparkExpression(spark, tableName, filter);

@rdblue rdblue Jan 9, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

collectResolvedSparkExpression is really expensive. Why does this need to call it?

Can't this produce Expression instances directly rather than building strings and converting with fake plans?

@huaxingao huaxingao Jan 10, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@rdblue Thank you very much for reviewing my PR on the weekend!

Do you mean constructing a filter Expression instead of letting Spark generate the Expression? I initially generated Expression like this

        BoundReference ref = new BoundReference(index, dataType, true);
        switch (dataType.typeName()) {
          case "integer":
            filterExpressions.add(new EqualTo(ref,
                org.apache.spark.sql.catalyst.expressions.Literal.create(Integer.parseInt(entry.getValue()),
                DataTypes.IntegerType)));
            break;

There are some concerns from the reviewers because we need to test each of the data types. Then I changed the code to call collectResolvedSparkExpression.

I will address all the other comments after I find out what to do for this one, so I can fix all the problems in one commit.

@rdblue rdblue Jan 10, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, I think you should construct the filter expression directly rather than calling collectResolvedSparkExpression.

@huaxingao huaxingao Jan 12, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@rdblue I changed the code to construct filter expression directly. Could you please take one more look? Thank you very much!

I have manually checked to make sure the filter expression can be created correctly for all the numeric types, Date and Timestamp. There are tests for String and int partition filters in TestAddFilesProcedure, and I added a test for Date partition filter. There is a bug with Timestamp partition filter in the current code. I will fix it in a separate PR.

The build failure doesn't seem to be related to my changes.

filterExpressions.add(expression);
} catch (AnalysisException e) {
throw new IllegalArgumentException("filter " + filter + " cannot be converted to Spark expression");

@rdblue rdblue Jan 9, 2022 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Minor: The exception message should follow the conventions for errors:

  • Use sentence case. That is, capitalize the first word of the message.
  • State what went wrong first, "Cannot convert filter to Spark"
  • Next, give context after a :, which in this case is the filter
  • Never swallow cause exceptions

This should be "throw new IllegalArgumentException("Cannot convert filter to Spark: " + filter, e)`

}
}
return filterExpressions;

@rdblue rdblue Jan 9, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit: missing whitespace between the control flow block and the following statement.

}

public static org.apache.spark.sql.catalyst.TableIdentifier toV1TableIdentifier(Identifier identifier) {
String[] namespace = identifier.namespace();

Expand Down
9 changes: 3 additions & 6 deletions spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/procedures/AddFilesProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -159,7 +159,8 @@ private static void ensureNameMappingPresent(Table table) {
private void importFileTable(Table table, Path tableLocation, String format, Map<String, String> partitionFilter,
boolean checkDuplicateFiles) {
// List Partitions via Spark InMemory file search interface
List<SparkPartition> partitions = Spark3Util.getPartitions(spark(), tableLocation, format);
List<SparkPartition> partitions =
Spark3Util.getPartitions(spark(), table.name(), tableLocation, format, partitionFilter);

if (table.spec().isUnpartitioned()) {
Preconditions.checkArgument(partitions.isEmpty(), "Cannot add partitioned files to an unpartitioned table");
Expand All @@ -172,11 +173,7 @@ private void importFileTable(Table table, Path tableLocation, String format, Map
} else {
Preconditions.checkArgument(!partitions.isEmpty(),
"Cannot find any partitions in table %s", partitions);

@kbendick kbendick Jan 6, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit / corner case:

Should we update the precondition message to indicate that it's possible that the filter didn't match any partitions? The current error message now might be kind of confusing to users if the file based table is partitioned. Maybe just Cannot find any matching partitions in table %s?

@huaxingao huaxingao Jan 7, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed. Thanks!

List<SparkPartition> filteredPartitions = SparkTableUtil.filterPartitions(partitions, partitionFilter);
Preconditions.checkArgument(!filteredPartitions.isEmpty(),
"Cannot find any partitions which match the given filter. Partition filter is %s",
MAP_JOINER.join(partitionFilter));
importPartitions(table, filteredPartitions, checkDuplicateFiles);
importPartitions(table, partitions, checkDuplicateFiles);
}
}

Expand Down