Skip to content

Commit 309da97

Browse files
committed
fix: ensure that null records are not included in record counts when determining if entity empty
1 parent 06eaee6 commit 309da97

5 files changed

Lines changed: 46 additions & 18 deletions

File tree

‎src/dve/core_engine/backends/implementations/duckdb/duckdb_helpers.py‎

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -110,9 +110,13 @@ def get_all_existing_ddb_tables(connection: DuckDBPyConnection) -> tuple[str]:
110110

111111
def relation_is_empty(relation: DuckDBPyRelation) -> bool:
112112
"""Check if a duckdb relation is empty"""
113-
if relation.limit(1).shape[0] > 0:
114-
return False
115-
return True
113+
if relation.limit(1).shape[0] == 0:
114+
return True
115+
116+
cl_all_null_check_exprs = [
117+
f'bool_and("{cl}" is NULL)' for cl in relation.columns if cl != RECORD_INDEX_COLUMN_NAME
118+
]
119+
return relation.aggregate(" AND ".join(cl_all_null_check_exprs)).fetchone()[0] # type: ignore
116120

117121

118122
def get_duckdb_type_from_annotation(type_annotation: Any) -> DuckDBPyType:
@@ -329,7 +333,7 @@ def duckdb_get_entity_count(cls):
329333

330334

331335
def _duckdb_check_entity_empty(self, entity: DuckDBPyRelation) -> bool: # pylint: disable=W0613
332-
return entity.shape[0] == 0
336+
return relation_is_empty(entity)
333337

334338

335339
def duckdb_check_entity_empty(cls):

‎src/dve/core_engine/backends/implementations/duckdb/rules.py‎

Lines changed: 6 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -372,18 +372,12 @@ def join_header(self, entities: DuckDBEntities, *, config: HeaderJoin) -> Messag
372372
),
373373
)
374374

375-
joined_rel = (
376-
source_rel
377-
.cross(target_rel)
378-
.select(
379-
StarExpression(
380-
exclude=[f"{config.target_name}.{c}" for c in target_rel.columns]
381-
),
382-
FunctionExpression(
383-
"struct_pack",
384-
*[ColumnExpression(f"{config.target_name}.{c}") for c in target_rel.columns],
385-
).alias(config.header_column_name)
386-
)
375+
joined_rel = source_rel.cross(target_rel).select(
376+
StarExpression(exclude=[f"{config.target_name}.{c}" for c in target_rel.columns]),
377+
FunctionExpression(
378+
"struct_pack",
379+
*[ColumnExpression(f"{config.target_name}.{c}") for c in target_rel.columns],
380+
).alias(config.header_column_name),
387381
)
388382

389383
entities[config.new_entity_name or config.entity_name] = joined_rel

‎src/dve/core_engine/backends/implementations/spark/spark_helpers.py‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -359,7 +359,17 @@ def object_to_spark_literal(obj: SparkLiteralType) -> Column:
359359

360360
def df_is_empty(dataframe: DataFrame) -> bool:
361361
"""Return whether a DataFrame is empty."""
362-
return not dataframe.select(lit(1)).head(1)
362+
363+
if not dataframe.select(lit(1)).head(1):
364+
return True
365+
366+
return (
367+
not dataframe.dropna(
368+
how="all", subset=[cl for cl in dataframe.columns if cl != RECORD_INDEX_COLUMN_NAME]
369+
)
370+
.select(lit(1))
371+
.head(1)
372+
)
363373

364374

365375
def _spark_read_parquet(self, path: URI, **kwargs) -> DataFrame:
@@ -457,7 +467,7 @@ def spark_get_entity_count(cls):
457467

458468

459469
def _spark_check_entity_empty(self, entity: DataFrame) -> bool: # pylint: disable=W0613
460-
return entity.count() == 0
470+
return df_is_empty(entity)
461471

462472

463473
def spark_check_entity_empty(cls):

‎tests/test_core_engine/test_backends/test_implementations/test_duckdb/test_duckdb_helpers.py‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -266,6 +266,12 @@ def test_relation_is_empty(temp_ddb_conn: DuckDBPyConnection):
266266
_, con = temp_ddb_conn
267267
rel = con.sql("SELECT 'abc' AS test").filter("test IS NULL")
268268
assert relation_is_empty(rel)
269+
270+
rel2 = con.sql("SELECT NULL as fld1, NULL as fld2 UNION ALL SELECT NULL as fld1, NULL as fld2")
271+
assert relation_is_empty(rel2)
272+
273+
rel3 = con.sql("SELECT 1 as fld1, NULL as fld2 UNION ALL SELECT NULL as fld1, NULL as fld2")
274+
assert not relation_is_empty(rel3)
269275

270276
@pytest.mark.parametrize("date_format, expected_regex", [("%Y-%m-%d", r"^[0-9]{4}\-[0-9]{2}\-[0-9]{2}$"),
271277
("%Y-%m-%dT%H:%M:%S", r"^[0-9]{4}\-[0-9]{2}\-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}$"),

‎tests/test_core_engine/test_backends/test_implementations/test_spark/test_spark_helpers.py‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
get_type_from_annotation,
3131
object_to_spark_literal,
3232
python_to_java_datetime_format,
33+
df_is_empty
3334
)
3435
from dve.metadata_parser.domain_types import conformatteddate
3536

@@ -354,3 +355,16 @@ def test_spark_filter_contract_errors(spark: SparkSession, example_data_contract
354355
("%Y-%m-%dT%H:%M:%S", "yyyy-MM-dd'T'HH:mm:ss")])
355356
def test_python_to_java_datetime_format(python_dt, expected_java_dt):
356357
assert python_to_java_datetime_format(python_dt) == expected_java_dt
358+
359+
360+
def test_df_is_empty(spark):
361+
df = spark.createDataFrame([], schema=StructType([StructField("col1", StringType()), StructField("col2", LongType())]))
362+
assert df_is_empty(df)
363+
df2 = spark.createDataFrame([(None, None), (None, None), (None, None)],
364+
schema=StructType([StructField("col1", StringType()), StructField("col2", LongType())]))
365+
assert df_is_empty(df2)
366+
df3 = spark.createDataFrame([(None, None), (None, None), (None, 3)],
367+
schema=StructType([StructField("col1", StringType()), StructField("col2", LongType())]))
368+
assert not df_is_empty(df3)
369+
370+

0 commit comments

Comments
 (0)