From 90dd4e2fd0a3458d9d729d4007cb21c39ffd22f6 Mon Sep 17 00:00:00 2001
From: keemgdeok <keemgdeok@gmail.com>
Date: Fri, 9 Oct 2026 16:41:36 +0900
Subject: [PATCH] Fix per-file null count consistency checks

---
 .../metadq/iceberg/EvidenceCollector.java     |  3 +
 .../metadq/iceberg/EvidenceCollectorTest.java | 84 +++++++++++++++++++
 2 files changed, 87 insertions(+)

diff --git a/src/main/java/io/github/keemgdeok/metadq/iceberg/EvidenceCollector.java b/src/main/java/io/github/keemgdeok/metadq/iceberg/EvidenceCollector.java
index f7f2ebf..26f52e4 100644
--- a/src/main/java/io/github/keemgdeok/metadq/iceberg/EvidenceCollector.java
+++ b/src/main/java/io/github/keemgdeok/metadq/iceberg/EvidenceCollector.java
@@ -193,6 +193,9 @@ && contains(file.upperBounds(), field.fieldId())) {
       if (isNegative(values) || isNegative(nulls) || isNegative(nans)) {
         inconsistent = true;
       }
+      if (nulls != null && nulls > file.recordCount()) {
+        inconsistent = true;
+      }
       if (values != null && nulls != null && nulls > values) {
         inconsistent = true;
       }
diff --git a/src/test/java/io/github/keemgdeok/metadq/iceberg/EvidenceCollectorTest.java b/src/test/java/io/github/keemgdeok/metadq/iceberg/EvidenceCollectorTest.java
index b72590f..9383891 100644
--- a/src/test/java/io/github/keemgdeok/metadq/iceberg/EvidenceCollectorTest.java
+++ b/src/test/java/io/github/keemgdeok/metadq/iceberg/EvidenceCollectorTest.java
@@ -5,6 +5,12 @@
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import io.github.keemgdeok.metadq.core.ReasonCode;
+import io.github.keemgdeok.metadq.core.RuleEvaluator;
+import io.github.keemgdeok.metadq.core.RuleSpec.NullRatioRule;
+import io.github.keemgdeok.metadq.core.RuleSpec.RowCountRule;
+import io.github.keemgdeok.metadq.core.Status;
+import java.time.Clock;
 import java.util.Map;
 import java.util.Set;
 import org.apache.iceberg.DataFiles;
@@ -14,6 +20,8 @@
 import org.apache.iceberg.exceptions.NotFoundException;
 import org.apache.iceberg.inmemory.InMemoryFileIO;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
 
 class EvidenceCollectorTest {
   @Test
@@ -114,6 +122,82 @@ void marksContradictoryCountsAsInconsistent() throws Exception {
     }
   }
 
+  @ParameterizedTest
+  @CsvSource({"10,20,false", "10,20,true", "0,1,false", "10,10,false", "10,0,false", "0,0,false"})
+  void checksNullCountsAgainstFileRecordCount(long rows, long nulls, boolean includeValueCount)
+      throws Exception {
+    try (DemoTableFactory.DemoTable demo = DemoTableFactory.createEmpty()) {
+      var file =
+          DataFiles.builder(demo.table().spec())
+              .withPath("memory://metadq/content/null-counts.parquet")
+              .withFormat(FileFormat.PARQUET)
+              .withFileSizeInBytes(128)
+              .withMetrics(
+                  new Metrics(
+                      rows,
+                      null,
+                      includeValueCount ? Map.of(2, Math.max(rows, nulls)) : null,
+                      Map.of(2, nulls),
+                      null))
+              .build();
+      demo.table().newAppend().appendFile(file).commit();
+
+      var evidence =
+          new EvidenceCollector().collect("demo.events", demo.table(), Set.of("user_id"));
+      boolean inconsistent = nulls > rows;
+      assertEquals(inconsistent, evidence.columns().get("user_id").inconsistentMetrics());
+      assertFalse(evidence.recordCountsInconsistent());
+      assertEquals(
+          Status.PASS,
+          RuleEvaluator.evaluate(new RowCountRule("rows", rows, rows), evidence, Clock.systemUTC())
+              .status());
+
+      var result =
+          RuleEvaluator.evaluate(
+              new NullRatioRule("nulls", "user_id", 1), evidence, Clock.systemUTC());
+      if (inconsistent) {
+        assertEquals(Status.UNKNOWN, result.status());
+        assertEquals(ReasonCode.INCONSISTENT_METRICS, result.reasonCode());
+      } else if (rows == 0) {
+        assertEquals(Status.UNKNOWN, result.status());
+        assertEquals(ReasonCode.EMPTY_TABLE, result.reasonCode());
+      } else {
+        assertEquals(Status.PASS, result.status());
+      }
+    }
+  }
+
+  @Test
+  void detectsFileNullCountContradictionHiddenByOtherFiles() throws Exception {
+    try (DemoTableFactory.DemoTable demo = DemoTableFactory.create()) {
+      var file =
+          DataFiles.builder(demo.table().spec())
+              .withPath("memory://metadq/content/invalid-score-nulls.parquet")
+              .withFormat(FileFormat.PARQUET)
+              .withFileSizeInBytes(128)
+              .withMetrics(new Metrics(10L, null, null, Map.of(3, 20L), null))
+              .build();
+      demo.table().newAppend().appendFile(file).commit();
+
+      var evidence = new EvidenceCollector().collect("demo.events", demo.table());
+
+      assertEquals(1_210, evidence.metadataRecordCount());
+      assertEquals(23, evidence.columns().get("score").nullCount());
+      assertTrue(evidence.columns().get("score").inconsistentMetrics());
+      assertFalse(evidence.columns().get("event_id").inconsistentMetrics());
+      assertEquals(
+          Status.PASS,
+          RuleEvaluator.evaluate(
+                  new NullRatioRule("event-nulls", "event_id", 0), evidence, Clock.systemUTC())
+              .status());
+      var result =
+          RuleEvaluator.evaluate(
+              new NullRatioRule("score-nulls", "score", 1), evidence, Clock.systemUTC());
+      assertEquals(Status.UNKNOWN, result.status());
+      assertEquals(ReasonCode.INCONSISTENT_METRICS, result.reasonCode());
+    }
+  }
+
   @Test
   void marksPositiveNullCountForRequiredColumnAsInconsistent() throws Exception {
     try (DemoTableFactory.DemoTable demo = DemoTableFactory.create()) {