Skip to content

Spark-3.0: Fix UnresolvedException for some filters in rewrite_data_files #3794

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 2 commits into from
Dec 30, 2021
Merged

Spark-3.0: Fix UnresolvedException for some filters in rewrite_data_files #3794

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5a08a5e
Spark-3.0: Fix UnresolvedException for some filters in rewrite_data_f…
ajantha-bhat Dec 23, 2021
d96ba25
Fix compile failure
ajantha-bhat Dec 23, 2021
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
159 changes: 117 additions & 42 deletions ...ions/src/test/java/org/apache/iceberg/spark/extensions/TestRewriteDataFilesProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,11 +64,11 @@ public void testRewriteDataFilesOnPartitionTable() {
List<Object[]> expectedRecords = currentData();

List<Object[]> output = sql(
"CALL %s.system.rewrite_data_files(table => '%s')", catalogName, tableIdent);
"CALL %s.system.rewrite_data_files(table => '%s')", catalogName, tableIdent);
Copy link
Contributor

@kbendick kbendick Dec 24, 2021 •
edited
Loading

Choose a reason for hiding this comment

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

Nit: Usually we try to avoid formatting only changes (even if they are the correct formatting), as the larger diff makes it harder for people that maintain forks to cherry-pick commits. If you aren’t changing the line, I’d suggest leaving it as is.

Same comment applies to all formatting only changes within a PR. 😊

Copy link
Member Author

@ajantha-bhat ajantha-bhat Dec 25, 2021

Choose a reason for hiding this comment

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

Agree. But I got a specific comment to fix indentation for this file in the spark-3.2 PR. Hence adopting it here also.

But I think checkstyle should catch over indented code in PR, so it can be caught and fixed in the respective PR itself. Let me explore why it is not catching


assertEquals("Action should rewrite 10 data files and add 2 data files (one per partition) ",
ImmutableList.of(row(10, 2)),
output);
ImmutableList.of(row(10, 2)),
output);

List<Object[]> actualRecords = currentData();
assertEquals("Data after compaction should not change", expectedRecords, actualRecords);
Expand All @@ -82,11 +82,11 @@ public void testRewriteDataFilesOnNonPartitionTable() {
List<Object[]> expectedRecords = currentData();

List<Object[]> output = sql(
"CALL %s.system.rewrite_data_files(table => '%s')", catalogName, tableIdent);
"CALL %s.system.rewrite_data_files(table => '%s')", catalogName, tableIdent);

assertEquals("Action should rewrite 10 data files and add 1 data files",
ImmutableList.of(row(10, 1)),
output);
ImmutableList.of(row(10, 1)),
output);

List<Object[]> actualRecords = currentData();
assertEquals("Data after compaction should not change", expectedRecords, actualRecords);
Expand All @@ -101,12 +101,12 @@ public void testRewriteDataFilesWithOptions() {

// set the min-input-files = 12, instead of default 5 to skip compacting the files.
List<Object[]> output = sql(
"CALL %s.system.rewrite_data_files(table => '%s', options => map('min-input-files','12'))",
catalogName, tableIdent);
"CALL %s.system.rewrite_data_files(table => '%s', options => map('min-input-files','12'))",
catalogName, tableIdent);

assertEquals("Action should rewrite 0 data files and add 0 data files",
ImmutableList.of(row(0, 0)),
output);
ImmutableList.of(row(0, 0)),
output);

List<Object[]> actualRecords = currentData();
assertEquals("Data should not change", expectedRecords, actualRecords);
Expand All @@ -121,13 +121,13 @@ public void testRewriteDataFilesWithSortStrategy() {

// set sort_order = c1 DESC LAST
List<Object[]> output = sql(
"CALL %s.system.rewrite_data_files(table => '%s', " +
"strategy => 'sort', sort_order => 'c1 DESC NULLS LAST')",
catalogName, tableIdent);
"CALL %s.system.rewrite_data_files(table => '%s', " +
"strategy => 'sort', sort_order => 'c1 DESC NULLS LAST')",
catalogName, tableIdent);

assertEquals("Action should rewrite 10 data files and add 1 data files",
ImmutableList.of(row(10, 1)),
output);
ImmutableList.of(row(10, 1)),
output);

List<Object[]> actualRecords = currentData();
assertEquals("Data after compaction should not change", expectedRecords, actualRecords);
Expand All @@ -142,12 +142,12 @@ public void testRewriteDataFilesWithFilter() {

// select only 5 files for compaction (files that may have c1 = 1)
List<Object[]> output = sql(
"CALL %s.system.rewrite_data_files(table => '%s'," +
" where => 'c1 = 1 and c2 is not null')", catalogName, tableIdent);
"CALL %s.system.rewrite_data_files(table => '%s'," +
" where => 'c1 = 1 and c2 is not null')", catalogName, tableIdent);

assertEquals("Action should rewrite 5 data files (containing c1 = 1) and add 1 data files",
ImmutableList.of(row(5, 1)),
output);
ImmutableList.of(row(5, 1)),
output);

List<Object[]> actualRecords = currentData();
assertEquals("Data after compaction should not change", expectedRecords, actualRecords);
Expand All @@ -166,14 +166,90 @@ public void testRewriteDataFilesWithFilterOnPartitionTable() {
" where => 'c2 = \"bar\"')", catalogName, tableIdent);

assertEquals("Action should rewrite 5 data files from single matching partition" +
"(containing c2 = bar) and add 1 data files",
"(containing c2 = bar) and add 1 data files",
ImmutableList.of(row(5, 1)),
output);

List<Object[]> actualRecords = currentData();
assertEquals("Data after compaction should not change", expectedRecords, actualRecords);
}

@Test
public void testRewriteDataFilesWithInFilterOnPartitionTable() {
createPartitionTable();
// create 5 files for each partition (c2 = 'foo' and c2 = 'bar')
insertData(10);
List<Object[]> expectedRecords = currentData();

// select only 5 files for compaction (files in the partition c2 in ('bar'))
List<Object[]> output = sql(
"CALL %s.system.rewrite_data_files(table => '%s'," +
" where => 'c2 in (\"bar\")')", catalogName, tableIdent);

assertEquals("Action should rewrite 5 data files from single matching partition" +
"(containing c2 = bar) and add 1 data files",
ImmutableList.of(row(5, 1)),
output);

List<Object[]> actualRecords = currentData();
assertEquals("Data after compaction should not change", expectedRecords, actualRecords);
}

@Test
public void testRewriteDataFilesWithAllPossibleFilters() {
createPartitionTable();
// create 5 files for each partition (c2 = 'foo' and c2 = 'bar')
insertData(10);

// Pass the literal value which is not present in the data files.
// So that parsing can be tested on a same dataset without actually compacting the files.

// EqualTo
sql("CALL %s.system.rewrite_data_files(table => '%s'," +
" where => 'c1 = 3')", catalogName, tableIdent);
// GreaterThan
sql("CALL %s.system.rewrite_data_files(table => '%s'," +
" where => 'c1 > 3')", catalogName, tableIdent);
// GreaterThanOrEqual
sql("CALL %s.system.rewrite_data_files(table => '%s'," +
" where => 'c1 >= 3')", catalogName, tableIdent);
// LessThan
sql("CALL %s.system.rewrite_data_files(table => '%s'," +
" where => 'c1 < 0')", catalogName, tableIdent);
// LessThanOrEqual
sql("CALL %s.system.rewrite_data_files(table => '%s'," +
" where => 'c1 <= 0')", catalogName, tableIdent);
// In
sql("CALL %s.system.rewrite_data_files(table => '%s'," +
" where => 'c1 in (3,4,5)')", catalogName, tableIdent);
// IsNull
sql("CALL %s.system.rewrite_data_files(table => '%s'," +
" where => 'c1 is null')", catalogName, tableIdent);
// IsNotNull
sql("CALL %s.system.rewrite_data_files(table => '%s'," +
" where => 'c3 is not null')", catalogName, tableIdent);
// And
sql("CALL %s.system.rewrite_data_files(table => '%s'," +
" where => 'c1 = 3 and c2 = \"bar\"')", catalogName, tableIdent);
// Or
sql("CALL %s.system.rewrite_data_files(table => '%s'," +
" where => 'c1 = 3 or c1 = 5')", catalogName, tableIdent);
// Not
sql("CALL %s.system.rewrite_data_files(table => '%s'," +
" where => 'c1 not in (1,2)')", catalogName, tableIdent);
// StringStartsWith
sql("CALL %s.system.rewrite_data_files(table => '%s'," +
" where => 'c2 like \"%s\"')", catalogName, tableIdent, "car%");

// TODO: Enable when org.apache.iceberg.spark.SparkFilters have implementations for StringEndsWith & StringContains
// StringEndsWith
// sql("CALL %s.system.rewrite_data_files(table => '%s'," +
// " where => 'c2 like \"%s\"')", catalogName, tableIdent, "%car");
// StringContains
// sql("CALL %s.system.rewrite_data_files(table => '%s'," +
// " where => 'c2 like \"%s\"')", catalogName, tableIdent, "%car%");
}

@Test
public void testRewriteDataFilesWithInvalidInputs() {
createTable();
Expand All @@ -182,41 +258,40 @@ public void testRewriteDataFilesWithInvalidInputs() {

// Test for invalid strategy
AssertHelpers.assertThrows("Should reject calls with unsupported strategy error message",
IllegalArgumentException.class, "unsupported strategy: temp. Only binpack,sort is supported",
() -> sql("CALL %s.system.rewrite_data_files(table => '%s', options => map('min-input-files','2'), " +
"strategy => 'temp')", catalogName, tableIdent));
IllegalArgumentException.class, "unsupported strategy: temp. Only binpack,sort is supported",
() -> sql("CALL %s.system.rewrite_data_files(table => '%s', options => map('min-input-files','2'), " +
"strategy => 'temp')", catalogName, tableIdent));

// Test for sort_order with binpack strategy
AssertHelpers.assertThrows("Should reject calls with error message",
IllegalArgumentException.class, "Cannot set strategy to sort, it has already been set",
() -> sql("CALL %s.system.rewrite_data_files(table => '%s', strategy => 'binpack', " +
"sort_order => 'c1 ASC NULLS FIRST')", catalogName, tableIdent));
IllegalArgumentException.class, "Cannot set strategy to sort, it has already been set",
() -> sql("CALL %s.system.rewrite_data_files(table => '%s', strategy => 'binpack', " +
"sort_order => 'c1 ASC NULLS FIRST')", catalogName, tableIdent));

// Test for sort_order with invalid null order
AssertHelpers.assertThrows("Should reject calls with error message",
IllegalArgumentException.class, "Unable to parse sortOrder:",
() -> sql("CALL %s.system.rewrite_data_files(table => '%s', strategy => 'sort', " +
"sort_order => 'c1 ASC none')", catalogName, tableIdent));
IllegalArgumentException.class, "Unable to parse sortOrder:",
() -> sql("CALL %s.system.rewrite_data_files(table => '%s', strategy => 'sort', " +
"sort_order => 'c1 ASC none')", catalogName, tableIdent));

// Test for sort_order with invalid sort direction
AssertHelpers.assertThrows("Should reject calls with error message",
IllegalArgumentException.class, "Unable to parse sortOrder:",
() -> sql("CALL %s.system.rewrite_data_files(table => '%s', strategy => 'sort', " +
"sort_order => 'c1 none NULLS FIRST')", catalogName, tableIdent));
IllegalArgumentException.class, "Unable to parse sortOrder:",
() -> sql("CALL %s.system.rewrite_data_files(table => '%s', strategy => 'sort', " +
"sort_order => 'c1 none NULLS FIRST')", catalogName, tableIdent));

// Test for sort_order with invalid column name
AssertHelpers.assertThrows("Should reject calls with error message",
ValidationException.class, "Cannot find field 'col1' in struct:" +
" struct<1: c1: optional int, 2: c2: optional string, 3: c3: optional string>",
() -> sql("CALL %s.system.rewrite_data_files(table => '%s', strategy => 'sort', " +
"sort_order => 'col1 DESC NULLS FIRST')", catalogName, tableIdent));
ValidationException.class, "Cannot find field 'col1' in struct:" +
" struct<1: c1: optional int, 2: c2: optional string, 3: c3: optional string>",
() -> sql("CALL %s.system.rewrite_data_files(table => '%s', strategy => 'sort', " +
"sort_order => 'col1 DESC NULLS FIRST')", catalogName, tableIdent));

// Test for sort_order with invalid filter column col1
AssertHelpers.assertThrows("Should reject calls with error message",
ValidationException.class, "Cannot find field 'col1' in struct:" +
" struct<1: c1: optional int, 2: c2: optional string, 3: c3: optional string>",
() -> sql("CALL %s.system.rewrite_data_files(table => '%s', " +
"where => 'col1 = 3')", catalogName, tableIdent));
IllegalArgumentException.class, "Cannot parse predicates in where option: col1 = 3",
() -> sql("CALL %s.system.rewrite_data_files(table => '%s', " +
"where => 'col1 = 3')", catalogName, tableIdent));
}

@Test
Expand Down Expand Up @@ -251,8 +326,8 @@ private void createPartitionTable() {
}

private void insertData(int filesCount) {
ThreeColumnRecord record1 = new ThreeColumnRecord(1, "foo", "detail1");
ThreeColumnRecord record2 = new ThreeColumnRecord(2, "bar", "detail2");
ThreeColumnRecord record1 = new ThreeColumnRecord(1, "foo", null);
ThreeColumnRecord record2 = new ThreeColumnRecord(2, "bar", null);

List<ThreeColumnRecord> records = Lists.newArrayList();
IntStream.range(0, filesCount / 2).forEach(i -> {
Expand Down
11 changes: 4 additions & 7 deletions ....0/spark/src/main/java/org/apache/iceberg/spark/procedures/RewriteDataFilesProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,8 +28,6 @@
import org.apache.spark.sql.AnalysisException;
import org.apache.spark.sql.catalyst.InternalRow;
import org.apache.spark.sql.catalyst.expressions.Expression;
import org.apache.spark.sql.catalyst.parser.ParseException;
import org.apache.spark.sql.catalyst.parser.ParserInterface;
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan;
import org.apache.spark.sql.catalyst.plans.logical.SetWriteDistributionAndOrdering;
import org.apache.spark.sql.catalyst.plans.logical.SortOrderParserUtil;
Expand Down Expand Up @@ -109,21 +107,20 @@ public InternalRow[] call(InternalRow args) {
}

String where = args.isNullAt(4) ? null : args.getString(4);
action = checkAndApplyFilter(action, where);
action = checkAndApplyFilter(action, where, table.name());

RewriteDataFiles.Result result = action.execute();

return toOutputRows(result);
});
}

private RewriteDataFiles checkAndApplyFilter(RewriteDataFiles action, String where) {
private RewriteDataFiles checkAndApplyFilter(RewriteDataFiles action, String where, String tableName) {
if (where != null) {
ParserInterface sqlParser = spark().sessionState().sqlParser();
try {
Expression expression = sqlParser.parseExpression(where);
Expression expression = SparkExpressionConverter.collectResolvedSparkExpression(spark(), tableName, where);
return action.filter(SparkExpressionConverter.convertToIcebergExpression(expression));
} catch (ParseException e) {
} catch (AnalysisException e) {
throw new IllegalArgumentException("Cannot parse predicates in where option: " + where);
}
}
Expand Down
17 changes: 17 additions & 0 deletions .../src/main/scala/org/apache/spark/sql/execution/datasources/SparkExpressionConverter.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,10 @@
package org.apache.spark.sql.execution.datasources

import org.apache.iceberg.spark.SparkFilters
import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.expressions.Expression
import org.apache.spark.sql.catalyst.plans.logical.Filter

object SparkExpressionConverter {

Expand All @@ -30,4 +33,18 @@ object SparkExpressionConverter {
// But these two conversions already exist and well tested. So, we are going with this approach.
SparkFilters.convert(DataSourceStrategy.translateFilter(sparkExpression, supportNestedPredicatePushdown = true).get)
}

@throws[AnalysisException]
def collectResolvedSparkExpression(session: SparkSession, tableName: String, where: String): Expression = {
var expression: Expression = null
// Add a dummy prefix linking to the table to collect the resolved spark expression from optimized plan.
val prefix = String.format("SELECT 42 from %s where ", tableName)
val logicalPlan = session.sessionState.sqlParser.parsePlan(prefix + where)
val optimizedLogicalPlan = session.sessionState.executePlan(logicalPlan).optimizedPlan
Copy link
Member Author

@ajantha-bhat ajantha-bhat Dec 23, 2021

Choose a reason for hiding this comment

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

The only change compared to spark-3.2 PR is this. executePlan accepts only one argument in spark-3.0 and spark-3.1

optimizedLogicalPlan.collectFirst {
case filter: Filter =>
expression = filter.expressions.head
}
expression
}
}