fix(datafusion): show an unreported partition statistic as NULL, not as -1

crates/integrations/datafusion/src/system_tables/partitions.rs

@@ -62,9 +62,12 @@ fn partitions_schema() -> SchemaRef {
         .get_or_init(|| {
             Arc::new(Schema::new(vec![
                 Field::new("partition", DataType::Utf8, true),
-                Field::new("record_count", DataType::Int64, false),
-                Field::new("file_size_in_bytes", DataType::Int64, false),
-                Field::new("file_count", DataType::Int64, false),
+                // Nullable because a statistic the catalog never had reported to it has no value to
+                // show. Declaring these non-null forced Partition::UNKNOWN to be rendered as -1,
+                // which reads as a measurement.
+                Field::new("record_count", DataType::Int64, true),
+                Field::new("file_size_in_bytes", DataType::Int64, true),
+                Field::new("file_count", DataType::Int64, true),
                 Field::new(
                     "last_update_time",
                     DataType::Timestamp(TimeUnit::Millisecond, None),
@@ -78,7 +81,7 @@ fn partitions_schema() -> SchemaRef {
                 Field::new("created_by", DataType::Utf8, true),
                 Field::new("updated_by", DataType::Utf8, true),
                 Field::new("options", DataType::Utf8, true),
-                Field::new("total_buckets", DataType::Int32, false),
+                Field::new("total_buckets", DataType::Int32, true),
                 Field::new("done", DataType::Boolean, false),
             ]))
         })
@@ -159,22 +162,25 @@ impl TableProvider for PartitionsTable {
 
         let n = rows.len();
         let mut partition_strings: Vec<Option<String>> = Vec::with_capacity(n);
-        let mut record_counts = Vec::with_capacity(n);
-        let mut file_sizes = Vec::with_capacity(n);
-        let mut file_counts = Vec::with_capacity(n);
+        let mut record_counts: Vec<Option<i64>> = Vec::with_capacity(n);
+        let mut file_sizes: Vec<Option<i64>> = Vec::with_capacity(n);
+        let mut file_counts: Vec<Option<i64>> = Vec::with_capacity(n);
         let mut last_update_times: Vec<Option<i64>> = Vec::with_capacity(n);
         let mut created_ats: Vec<Option<i64>> = Vec::with_capacity(n);
         let mut created_bys: Vec<Option<String>> = Vec::with_capacity(n);
         let mut updated_bys: Vec<Option<String>> = Vec::with_capacity(n);
         let mut options_jsons: Vec<Option<String>> = Vec::with_capacity(n);
-        let mut total_buckets = Vec::with_capacity(n);
+        let mut total_buckets: Vec<Option<i32>> = Vec::with_capacity(n);
         let mut dones = Vec::with_capacity(n);
 
         for (s, p) in rows {
             partition_strings.push(Some(s));
-            record_counts.push(p.record_count);
-            file_sizes.push(p.file_size_in_bytes);
-            file_counts.push(p.file_count);
+            // A field nobody reported on shows as NULL. Passing the placeholder straight through
+            // would claim the partition holds -1 rows, and 0 would claim it is empty.
+            record_counts.push(Partition::is_known(p.record_count).then_some(p.record_count));
+            file_sizes
+                .push(Partition::is_known(p.file_size_in_bytes).then_some(p.file_size_in_bytes));
+            file_counts.push(Partition::is_known(p.file_count).then_some(p.file_count));
             // 0 marks "no creation_time on any file"; real wall-clock is never
             // <= 0 in practice, so this never nullifies a genuine timestamp.
             last_update_times.push(if p.last_file_creation_time > 0 {
@@ -197,7 +203,8 @@ impl TableProvider for PartitionsTable {
                     })
                     .transpose()?,
             );
-            total_buckets.push(p.total_buckets);
+            total_buckets
+                .push(Some(p.total_buckets).filter(|b| *b != Partition::UNKNOWN_TOTAL_BUCKETS));
             dones.push(p.done);
         }
 

crates/integrations/datafusion/tests/sql_context_tests.rs

@@ -152,6 +152,7 @@ impl Catalog for MetadataListingCatalog {
 struct PartitionCatalog {
     inner: Arc<FileSystemCatalog>,
     fail_list_partitions: AtomicBool,
+    unknown_statistics: AtomicBool,
     partition_identifiers: Mutex<Vec<Identifier>>,
 }
 
@@ -160,6 +161,7 @@ impl PartitionCatalog {
         Self {
             inner,
             fail_list_partitions: AtomicBool::new(false),
+            unknown_statistics: AtomicBool::new(false),
             partition_identifiers: Mutex::new(Vec::new()),
         }
     }
@@ -168,6 +170,12 @@ impl PartitionCatalog {
         self.fail_list_partitions.store(fail, Ordering::SeqCst);
     }
 
+    /// Report every statistic as never measured, the way a catalog does for a partition that was
+    /// registered but never had statistics reported to it.
+    fn set_unknown_statistics(&self, unknown: bool) {
+        self.unknown_statistics.store(unknown, Ordering::SeqCst);
+    }
+
     fn take_partition_identifiers(&self) -> Vec<Identifier> {
         std::mem::take(&mut *self.partition_identifiers.lock().unwrap())
     }
@@ -266,7 +274,17 @@ impl Catalog for PartitionCatalog {
             .push(identifier.clone());
 
         let Some(branch) = identifier.branch_name()? else {
-            return self.inner.list_partitions(identifier).await;
+            let mut partitions = self.inner.list_partitions(identifier).await?;
+            if self.unknown_statistics.load(Ordering::SeqCst) {
+                for partition in &mut partitions {
+                    partition.record_count = paimon::spec::Partition::UNKNOWN;
+                    partition.file_size_in_bytes = paimon::spec::Partition::UNKNOWN;
+                    partition.file_count = paimon::spec::Partition::UNKNOWN;
+                    partition.last_file_creation_time = paimon::spec::Partition::UNKNOWN;
+                    partition.total_buckets = paimon::spec::Partition::UNKNOWN_TOTAL_BUCKETS;
+                }
+            }
+            return Ok(partitions);
         };
         if self.fail_list_partitions.load(Ordering::SeqCst) {
             return Err(paimon::Error::Unsupported {
@@ -580,6 +598,78 @@ async fn test_select_branch_table_reads_branch_snapshot() {
     .await;
 }
 
+/// A statistic the catalog never had reported to it has to read as NULL.
+///
+/// The alternative is what this used to do: declare the columns non-nullable and let
+/// `Partition::UNKNOWN` through, so `$partitions` claimed the partition holds -1 rows and -1 files.
+/// Reporting it as `0` instead would be worse still — that is a real measurement meaning empty.
+#[tokio::test]
+async fn test_partitions_system_table_shows_unreported_statistics_as_null() {
+    let (_tmp, file_catalog) = create_test_env();
+    let catalog = Arc::new(PartitionCatalog::new(file_catalog.clone()));
+    let mut sql_context = SQLContext::new();
+    sql_context
+        .register_catalog("paimon", catalog.clone())
+        .await
+        .unwrap();
+
+    sql_context
+        .sql(
+            "CREATE TABLE paimon.default.unknown_stats_orders \
+             (id INT, name STRING) PARTITIONED BY (id)",
+        )
+        .await
+        .unwrap()
+        .collect()
+        .await
+        .unwrap();
+    sql_context
+        .sql("INSERT INTO paimon.default.unknown_stats_orders VALUES (1, 'a')")
+        .await
+        .unwrap()
+        .collect()
+        .await
+        .unwrap();
+
+    let sql = "SELECT record_count, file_size_in_bytes, file_count, total_buckets \
+               FROM paimon.default.unknown_stats_orders$partitions";
+
+    // Measured statistics still arrive as values, so a NULL below means unknown and not that the
+    // column stopped being populated at all.
+    let measured = sql_context.sql(sql).await.unwrap().collect().await.unwrap();
+    let measured = &measured[0];
+    assert_eq!(measured.num_rows(), 1);
+    for column in 0..4 {
+        assert!(
+            !measured.column(column).is_null(0),
+            "column {column} should carry a measurement before the switch"
+        );
+    }
+
+    catalog.set_unknown_statistics(true);
+
+    let batches = sql_context.sql(sql).await.unwrap().collect().await.unwrap();
+    let batch = &batches[0];
+    assert_eq!(batch.num_rows(), 1);
+    for column in 0..4 {
+        assert!(
+            batch.column(column).is_null(0),
+            "column {column} was never measured and must read as NULL"
+        );
+    }
+
+    // The partition itself is still registered; only its statistics are unknown.
+    assert_eq!(
+        collect_string_column(
+            &sql_context,
+            "SELECT \"partition\" FROM paimon.default.unknown_stats_orders$partitions",
+            "partition",
+        )
+        .await,
+        vec!["id=1".to_string()]
+    );
+}
+
 #[tokio::test]
 async fn test_branch_partitions_system_table_reads_branch_snapshot() {
     let (_tmp, file_catalog) = create_test_env();

crates/paimon/src/spec/partition.rs

@@ -46,6 +46,27 @@ pub struct Partition {
     pub options: Option<HashMap<String, String>>,
 }
 
+impl Partition {
+    /// A statistic the catalog never had reported to it.
+    ///
+    /// On this plane — what `list_partitions` observed — any negative value means the field was
+    /// never measured, and `-1` is the canonical spelling. `0` is an exact zero and must stay
+    /// distinguishable from it: a partition nobody reported on is not an empty partition. The
+    /// counterpart for file-level counts is [`crate::spec::DataFileMeta::ROW_COUNT_UNKNOWN`].
+    ///
+    /// This says nothing about the delta plane used by snapshot commits, where a negative value is
+    /// a decrement to apply rather than a missing measurement.
+    pub const UNKNOWN: i64 = -1;
+
+    /// A partition with no buckets, or none the catalog knows of. Format tables have no buckets.
+    pub const UNKNOWN_TOTAL_BUCKETS: i32 = -1;
+
+    /// Whether a statistic read off a partition is a measurement rather than [`Self::UNKNOWN`].
+    pub fn is_known(value: i64) -> bool {
+        value >= 0
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -108,4 +129,17 @@ mod tests {
         let decoded: Partition = serde_json::from_str(json).unwrap();
         assert_eq!(decoded.spec.get("dt"), Some(&"2024-01-01".to_string()));
     }
+
+    #[test]
+    fn test_unknown_is_negative_and_zero_is_a_measurement() {
+        assert_eq!(Partition::UNKNOWN, -1);
+        assert_eq!(Partition::UNKNOWN_TOTAL_BUCKETS, -1);
+
+        assert!(!Partition::is_known(Partition::UNKNOWN));
+        assert!(!Partition::is_known(-42));
+
+        // A partition nobody reported on is not an empty partition.
+        assert!(Partition::is_known(0));
+        assert!(Partition::is_known(1));
+    }
 }