Skip to content
Merged
Show file tree
Hide file tree
Changes from 5 commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -162,6 +162,8 @@ abstract class DeltaProviderBase extends DeltaIOProvider {
dvFilterInput.copy(projectList = inputList.filterNot(_.name == "_metadata"))
.withNewChildren(Seq(
fsse.copy(
originalOutput =
fsse.originalOutput.filterNot(_.name == "_tmp_metadata_row_index"),
requiredSchema = StructType(
fsse.requiredSchema.filterNot(_.name == "_tmp_metadata_row_index")
))(fsse.rapidsConf)))))))
Expand Down
19 changes: 19 additions & 0 deletions integration_tests/src/main/python/delta_lake_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -160,3 +160,22 @@ def convert_and_setup_name_mapping(spark):
with_cpu_session(setup_parquet_table, {"spark.sql.parquet.fieldId.write.enabled": str(enable_deletion_vectors).lower()})
with_cpu_session(convert_and_setup_name_mapping, conf={"spark.databricks.delta.properties.defaults.enableDeletionVectors": "false"})
assert_gpu_and_cpu_are_equal_collect(lambda spark: spark.read.format("delta").load(data_path))

@allow_non_gpu(*delta_meta_allow)
@delta_lake
@ignore_order(local=True)
@pytest.mark.skipif(not is_spark_340_or_later(), reason="Deletion Vectors only supported on Spark 3.4.0+")
def test_delta_filter_out_metadata_col(spark_tmp_path):
data_path = spark_tmp_path + "/DELTA_DATA"

def create_delta(spark):
two_col_df(spark, int_gen, int_gen).coalesce(1).write.format("delta") \

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

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

Is int_gen guaranteed to have a 0 that you are later deleting?

Is coalesce(1) the same as passing num_slices=1 to two_col_df ?

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

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

0 is guaranteed to be present as it's a special_case in the IntegerGen

.option("delta.enableDeletionVectors", "true") \
.partitionBy("a").save(data_path)

spark.sql(f"DELETE FROM delta.`{data_path}` WHERE b = 0")

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

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

Should we assert that we deleted at least one row by looking at the result df returned by this statement?

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

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

good idea. I think we should


def read_table(spark):
return spark.sql(f"SELECT * FROM delta.`{data_path}` ")

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

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

Alternatively to the above suggestions, can we assert the explain output includes DV-enabled scan with __delta_internal_is_row_deleted column?

with_cpu_session(create_delta)
assert_gpu_and_cpu_are_equal_collect(read_table)