fix: backport scan statistics and partitioning fixes to release/5e02efa9 (#9783)

* fix: preserve count correctness after repartition (#9154) * fix: preserve count correctness after repartition Signed-off-by: WenyXu <wenymedia@gmail.com> * fix: restore full predicate guard for count statistics Signed-off-by: WenyXu <wenymedia@gmail.com> * test: update count plans for full predicate guard Signed-off-by: WenyXu <wenymedia@gmail.com> * fix: preserve count statistics for safe partition scans Signed-off-by: WenyXu <wenymedia@gmail.com> * test: retain repartition home guard and cover staging flush Signed-off-by: WenyXu <wenymedia@gmail.com> --------- Signed-off-by: WenyXu <wenymedia@gmail.com> (cherry picked from commit d9b97796a6) Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> * fix: only push down aggregates grouped by the partition columns themselves (#9337) * fix: only push down aggregates grouped by the partition columns themselves Signed-off-by: Dennis Zhuang <killme2008@gmail.com> * fix: keep grouping sets on the frontend for partitioned tables Signed-off-by: Dennis Zhuang <killme2008@gmail.com> --------- Signed-off-by: Dennis Zhuang <killme2008@gmail.com> (cherry picked from commit def0c2ec5c) Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> * fix(table): require complete primary keys for hash partitioning (#9507) Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> (cherry picked from commit 28ec3ff5aa) Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> * fix(table): stop claiming exact null and distinct counts for append scans (#9593) RegionScanExec reported null_count = 0 and distinct_count = total_rows for every column of an append-only table. DataFusion answers count(col) and count(DISTINCT col) from these statistics, so both returned the row count. Keep the exact row count and an exact zero null count for non-nullable columns, so count(*) is still answered from statistics. Signed-off-by: Dennis Zhuang <killme2008@gmail.com> (cherry picked from commit 5cfdc30a69) Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> * fix(table): stop claiming tag order for logical metric scans (#9522) * test: cover metric logical scan tag ordering Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * fix(table): only claim projected primary key prefix ordering Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * fix(table): keep physical key ordering for metric projections Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test(table): compare unknown partition counts explicitly Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test(table): model scoped internal TSID scan properties Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --------- Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> (cherry picked from commit 75bc4ce46d) Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> * test: regenerate append_count_nulls result for release plan rendering Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> --------- Signed-off-by: WenyXu <wenymedia@gmail.com> Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> Signed-off-by: Dennis Zhuang <killme2008@gmail.com> Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> Co-authored-by: Weny Xu <wenymedia@gmail.com> Co-authored-by: dennis zhuang <killme2008@gmail.com> Co-authored-by: discord9 <discord9@163.com>
2026-10-27 14:16:18 +00:00 · 2026-10-20 08:32:32 -07:00
parent 1e5fef5809
commit fd3faeb7ed
16 changed files with 1442 additions and 45 deletions
@@ -1688,6 +1688,27 @@ impl ScanInput {
        );
    }

+    /// Whether physical source row counts equal the rows visible to this region,
+    /// before applying query predicates.
+    pub(crate) fn total_rows_is_exact(&self) -> bool {
+        if self.region_partition_expr.is_none() {
+            return true;
+        }
+
+        // Extension ranges do not expose their partition expressions.
+        #[cfg(feature = "enterprise")]
+        if !self.extension_ranges.is_empty() {
+            return false;
+        }
+
+        // Repartition flushes existing memtables before entering staging. New
+        // writes are routed under the target partition rules, whereas historical
+        // SSTs can be shared by regions and still need partition filtering.
+        self.files
+            .iter()
+            .all(|file| self.should_skip_region_partition(file))
+    }
+
    pub(crate) fn total_rows(&self) -> usize {
        let rows_in_files: usize = self.files.iter().map(|f| f.num_rows()).sum();
        let rows_in_memtables: usize = self.memtables.iter().map(|m| m.stats().num_rows()).sum();
@@ -2325,6 +2346,43 @@ mod tests {
            .with_files(vec![file])
    }

+    #[tokio::test]
+    async fn test_total_rows_is_exact_after_partition_filter() {
+        let expr = partition_col("k0").gt_eq(Value::String("foo".into()));
+        let other = partition_col("k0").gt_eq(Value::String("bar".into()));
+        for (region_expr, file_exprs, exact) in [
+            (None, vec![None], true),
+            (None, vec![Some(expr.clone())], true),
+            (Some(expr.clone()), vec![], true),
+            (Some(expr.clone()), vec![Some(expr.clone())], true),
+            (Some(expr.clone()), vec![None], false),
+            (Some(expr.clone()), vec![Some(other)], false),
+            (Some(expr.clone()), vec![Some(expr), None], false),
+        ] {
+            let mut builder =
+                RegionMetadataBuilder::from_existing(metadata_with_primary_key(vec![0, 1], false));
+            builder.partition_expr_json(region_expr.map(|expr| expr.as_json_str().unwrap()));
+            let metadata = Arc::new(builder.build_without_validation().unwrap());
+            let files = file_exprs
+                .into_iter()
+                .map(|partition_expr| {
+                    FileHandle::new(
+                        FileMeta {
+                            partition_expr,
+                            ..Default::default()
+                        },
+                        Arc::new(crate::sst::file_purger::NoopFilePurger),
+                    )
+                })
+                .collect();
+            let input = new_scan_input(metadata, vec![])
+                .await
+                .with_files(files)
+                .build();
+            assert_eq!(input.total_rows_is_exact(), exact);
+        }
+    }
+
    /// Helper to create a timestamp millisecond literal.
    fn ts_lit(val: i64) -> datafusion_expr::Expr {
        lit(ScalarValue::TimestampMillisecond(Some(val), None))
@@ -76,7 +76,8 @@ impl SeqScan {
    pub(crate) fn new(input: ScanInput) -> Self {
        let mut properties = ScannerProperties::default()
            .with_append_mode(input.append_mode)
-            .with_total_rows(input.total_rows());
+            .with_total_rows(input.total_rows())
+            .with_total_rows_is_exact(input.append_mode && input.total_rows_is_exact());
        if let Some(counters) = input.query_stat_counters.clone() {
            properties.set_query_stat_counters(counters);
        }
@@ -147,7 +147,8 @@ impl SeriesScan {
        };
        let mut properties = ScannerProperties::default()
            .with_append_mode(input.append_mode)
-            .with_total_rows(input.total_rows());
+            .with_total_rows(input.total_rows())
+            .with_total_rows_is_exact(input.append_mode && input.total_rows_is_exact());
        if let Some(counters) = input.query_stat_counters.clone() {
            properties.set_query_stat_counters(counters);
        }
@@ -61,7 +61,8 @@ impl UnorderedScan {
    pub(crate) fn new(input: ScanInput) -> Self {
        let mut properties = ScannerProperties::default()
            .with_append_mode(input.append_mode)
-            .with_total_rows(input.total_rows());
+            .with_total_rows(input.total_rows())
+            .with_total_rows_is_exact(input.append_mode && input.total_rows_is_exact());
        if let Some(counters) = input.query_stat_counters.clone() {
            properties.set_query_stat_counters(counters);
        }
@@ -38,6 +38,7 @@ use datafusion_expr::{
 };
 use datafusion_functions::datetime::date_bin;
 use datafusion_functions::datetime::expr_fn::now;
+use datafusion_functions::unicode::expr_fn::substring;
 use datafusion_sql::TableReference;
 use datatypes::data_type::ConcreteDataType;
 use datatypes::schema::{ColumnSchema, SchemaBuilder, SchemaRef};
@@ -1094,6 +1095,40 @@ fn expand_proj_alias_aliased_part_col_aggr() {
    assert_eq!(expected, result.to_string());
 }

+/// `substr(pk1, 1, 1)` maps rows from different partitions to the same group, so the
+/// aggregate has to be merged on the frontend even though it references `pk1`.
+#[test]
+fn expand_expr_over_part_col_aggr() {
+    init_default_ut_logging();
+    let test_table = TestTable::table_with_name(0, "t".to_string());
+    let table_source = Arc::new(DefaultTableSource::new(Arc::new(
+        DfTableProviderAdapter::new(test_table),
+    )));
+    let plan = LogicalPlanBuilder::scan_with_filters("t", table_source, None, vec![])
+        .unwrap()
+        .aggregate(
+            vec![substring(col("pk1"), lit(1i64), lit(1i64)), col("pk2")],
+            vec![min(col("number"))],
+        )
+        .unwrap()
+        .build()
+        .unwrap();
+
+    let config = ConfigOptions::default();
+    let result = DistPlannerAnalyzer {}.analyze(plan, &config).unwrap();
+
+    let expected = [
+        "Projection: substr(t.pk1,Int64(1),Int64(1)), t.pk2, min(t.number)",
+        "  Aggregate: groupBy=[[substr(t.pk1,Int64(1),Int64(1)), t.pk2]], aggr=[[__min_merge(__min_state(t.number)) AS min(t.number)]]",
+        "    MergeScan [is_placeholder=false, remote_input=[",
+        "Aggregate: groupBy=[[substr(t.pk1, Int64(1), Int64(1)), t.pk2]], aggr=[[__min_state(t.number)]]",
+        "  TableScan: t",
+        "]]",
+    ]
+    .join("\n");
+    assert_eq!(expected, result.to_string());
+}
+
 /// notice that step aggr then part col aggr seems impossible as the partition columns for part col aggr
 /// can't pass through the step aggr without making step aggr also a part col aggr
 /// so here only test part col aggr -> step aggr case
@@ -156,7 +156,14 @@ impl Categorizer {
            LogicalPlan::Filter(filter) => Self::check_expr(&filter.predicate),
            LogicalPlan::Window(_) => Commutativity::Unimplemented,
            LogicalPlan::Aggregate(aggr) => {
-                let is_all_steppable = is_all_aggr_exprs_steppable(&aggr.aggr_expr);
+                // The state/merge split maps each group expression to one output column,
+                // which doesn't hold for grouping sets.
+                let has_grouping_set = aggr
+                    .group_expr
+                    .iter()
+                    .any(|expr| matches!(expr, Expr::GroupingSet(_)));
+                let is_all_steppable =
+                    !has_grouping_set && is_all_aggr_exprs_steppable(&aggr.aggr_expr);
                let matches_partition = Self::check_partition(&aggr.group_expr, &partition_cols);
                if !matches_partition && is_all_steppable {
                    debug!("Plan is steppable: {plan}");
@@ -345,26 +352,34 @@ impl Categorizer {
    /// Return true if the given expr and partition cols satisfied the rule.
    /// In this case the plan can be treated as fully commutative.
    ///
-    /// So only if all partition columns show up in `exprs`, return true.
+    /// So only if every partition column is itself one of `exprs`, return true.
    /// Otherwise return false.
    ///
+    /// An expression that only references a partition column, like `substr(host, 3, 1)`
+    /// or `k % 2`, doesn't count: it can put rows from different partitions into the same
+    /// group.
    fn check_partition(exprs: &[Expr], partition_cols: &AliasMapping) -> bool {
-        let mut ref_cols = HashSet::new();
-        for expr in exprs {
-            expr.add_column_refs(&mut ref_cols);
-        }
-        let ref_cols = ref_cols
-            .into_iter()
-            .map(|c| c.name.clone())
+        let group_cols = exprs
+            .iter()
+            .filter_map(|expr| {
+                let mut expr = expr;
+                while let Expr::Alias(alias) = expr {
+                    expr = &alias.expr;
+                }
+                match expr {
+                    Expr::Column(column) => Some(column.name.clone()),
+                    _ => None,
+                }
+            })
            .collect::<HashSet<_>>();
        for all_alias in partition_cols.values() {
            let all_alias = all_alias
                .iter()
                .map(|c| c.name.clone())
                .collect::<HashSet<_>>();
-            // check if ref columns intersect with all alias of partition columns
+            // check if group columns intersect with all alias of partition columns
            // is empty, if it's empty, not all partition columns show up in `exprs`
-            if ref_cols.intersection(&all_alias).count() == 0 {
+            if group_cols.intersection(&all_alias).count() == 0 {
                return false;
            }
        }
@@ -306,6 +306,9 @@ pub struct ScannerProperties {
    /// [ScannerProperties::append_mode] is true.
    total_rows: usize,

+    /// Whether total_rows is exact before applying query predicates.
+    total_rows_is_exact: bool,
+
    /// Whether to yield an empty batch to distinguish partition ranges.
    pub distinguish_partition_range: bool,

@@ -334,12 +337,19 @@ impl ScannerProperties {
        self
    }

+    /// Sets whether the source row count is exact before query predicates.
+    pub fn with_total_rows_is_exact(mut self, exact: bool) -> Self {
+        self.total_rows_is_exact = exact;
+        self
+    }
+
    /// Creates a new [`ScannerProperties`] with the given partitioning.
    pub fn new(partitions: Vec<Vec<PartitionRange>>, append_mode: bool, total_rows: usize) -> Self {
        Self {
            partitions,
            append_mode,
            total_rows,
+            total_rows_is_exact: false,
            distinguish_partition_range: false,
            target_partitions: 0,
            logical_region: false,
@@ -374,6 +384,11 @@ impl ScannerProperties {
        self.total_rows
    }

+    /// Returns whether the source row count is exact before query predicates.
+    pub fn total_rows_is_exact(&self) -> bool {
+        self.total_rows_is_exact
+    }
+
    /// Returns whether the scanner is scanning a logical region.
    pub fn is_logical_region(&self) -> bool {
        self.logical_region
@@ -118,21 +118,34 @@ impl RegionScanExec {
        }

        let metadata = scanner.metadata();
-        let mut pk_names = metadata
+        // Keep the primary key columns in the order reported by the metadata (for a metric
+        // region that is the physical row key order). Reordering them would let a projected
+        // logical tag look like a prefix of the row key, which yields ordering and
+        // partitioning claims the scanner cannot honour.
+        let pk_names = metadata
            .primary_key_columns()
            .map(|col| col.column_schema.name.clone())
            .collect::<Vec<_>>();
-        // workaround for logical table
-        if scanner.properties().is_logical_region() {
-            pk_names.sort_unstable();
-        }
+        // A region emits rows ordered by its primary key columns and then by the time
+        // index column. A primary key column that is not projected is absent from the
+        // arrow schema, and it also breaks the ordering of every primary key column that
+        // follows it (e.g. the physical `__table_id`/`__tsid` prefix of a metric region
+        // when reading a logical table). Thus only the longest fully projected prefix of
+        // the primary key can describe the output ordering, and that prefix can be empty.
+        let projected_pk_prefix_len = pk_names
+            .iter()
+            .position(|name| Column::new_with_schema(name, &arrow_schema).is_err())
+            .unwrap_or(pk_names.len());
+        // Whether the projected prefix covers the whole row key.
+        let all_pk_projected = projected_pk_prefix_len == pk_names.len();
+        let projected_pk_names = &pk_names[..projected_pk_prefix_len];
        let pk_columns = pk_names
            .iter()
            .filter_map(
                |col| Some(Arc::new(Column::new_with_schema(col, &arrow_schema).ok()?) as _),
            )
            .collect::<Vec<_>>();
-        let pk_sort_columns: Vec<PhysicalSortExpr> = pk_names
+        let pk_sort_columns: Vec<PhysicalSortExpr> = projected_pk_names
            .iter()
            .filter_map(|col| {
                Some(PhysicalSortExpr::new(
@@ -165,7 +178,9 @@ impl RegionScanExec {
                let mut orderings = Vec::with_capacity(2);

                let mut pk_time_ordering = pk_sort_columns.clone();
-                if let Some(ts) = ts_col.clone() {
+                // The time index column extends the ordering only when the whole row key
+                // precedes it; otherwise rows of different series can interleave.
+                if all_pk_projected && let Some(ts) = ts_col.clone() {
                    pk_time_ordering.push(ts);
                }
                orderings.push(pk_time_ordering);
@@ -213,7 +228,10 @@ impl RegionScanExec {
                    Ok(tsid_col) => {
                        Partitioning::Hash(vec![Arc::new(tsid_col) as _], num_output_partition)
                    }
-                    Err(_) => Partitioning::Hash(pk_columns.clone(), num_output_partition),
+                    Err(_) if pk_columns.len() == pk_names.len() => {
+                        Partitioning::Hash(pk_columns.clone(), num_output_partition)
+                    }
+                    Err(_) => Partitioning::UnknownPartitioning(num_output_partition),
                }
            }
            Some(TimeSeriesDistribution::TimeWindowed) | None => {
@@ -482,30 +500,40 @@ impl ExecutionPlan for RegionScanExec {
    }

    fn partition_statistics(&self, partition: Option<usize>) -> DfResult<Statistics> {
-        if partition.is_some() {
+        if partition.is_some() || !self.append_mode {
            return Ok(Statistics::new_unknown(self.schema().as_ref()));
        }

-        let statistics =
-            if self.append_mode && !self.scanner.lock().unwrap().has_predicate_without_region() {
-                let column_statistics = self
-                    .arrow_schema
-                    .fields
-                    .iter()
-                    .map(|_| ColumnStatistics {
-                        distinct_count: Precision::Exact(self.total_rows),
-                        null_count: Precision::Exact(0), // all null rows are counted for append-only table
-                        ..Default::default()
-                    })
-                    .collect();
-                Statistics {
-                    num_rows: Precision::Exact(self.total_rows),
-                    total_byte_size: Default::default(),
-                    column_statistics,
-                }
-            } else {
-                Statistics::new_unknown(&self.arrow_schema)
-            };
+        let scanner = self.scanner.lock().unwrap();
+        let statistics = if scanner.properties().total_rows_is_exact()
+            && !scanner.has_predicate_without_region()
+        {
+            // Only the row count is known. DataFusion answers `count(col)` as
+            // `num_rows - null_count` and `count(DISTINCT col)` as `distinct_count`
+            // without scanning, so these must stay absent unless the schema
+            // guarantees them. Non-nullable columns (including the time index that
+            // `count(*)` is rewritten to) keep an exact zero null count.
+            let column_statistics = self
+                .arrow_schema
+                .fields
+                .iter()
+                .map(|field| ColumnStatistics {
+                    null_count: if field.is_nullable() {
+                        Precision::Absent
+                    } else {
+                        Precision::Exact(0)
+                    },
+                    ..Default::default()
+                })
+                .collect();
+            Statistics {
+                num_rows: Precision::Exact(self.total_rows),
+                total_byte_size: Default::default(),
+                column_statistics,
+            }
+        } else {
+            Statistics::new_unknown(&self.arrow_schema)
+        };
        Ok(statistics)
    }

@@ -679,10 +707,12 @@ mod test {
    use datatypes::vectors::{Int32Vector, TimestampMillisecondVector, VectorRef};
    use futures::TryStreamExt;
    use store_api::metadata::{ColumnMetadata, RegionMetadataBuilder};
+    use store_api::metric_engine_consts::DATA_SCHEMA_TABLE_ID_COLUMN_NAME;
    use store_api::region_engine::{
        PrepareRequest, QueryScanContext, RegionScanner, ScannerProperties, SinglePartitionScanner,
    };
-    use store_api::storage::RegionId;
+    use store_api::storage::consts::ReservedColumnId;
+    use store_api::storage::{RegionId, TimeSeriesDistribution};

    use super::*;

@@ -845,6 +875,377 @@ mod test {
        )
    }

+    #[test]
+    fn test_per_series_partitioning_requires_all_primary_keys() {
+        let full_schema = Arc::new(Schema::new(vec![
+            ColumnSchema::new("marker", ConcreteDataType::int32_datatype(), false),
+            ColumnSchema::new("host", ConcreteDataType::string_datatype(), false),
+            ColumnSchema::new(
+                "ts",
+                ConcreteDataType::timestamp_millisecond_datatype(),
+                false,
+            ),
+        ]));
+        let mut builder = RegionMetadataBuilder::new(RegionId::new(1234, 5683));
+        builder
+            .push_column_metadata(ColumnMetadata {
+                column_schema: ColumnSchema::new(
+                    "marker",
+                    ConcreteDataType::int32_datatype(),
+                    false,
+                ),
+                semantic_type: SemanticType::Tag,
+                column_id: 1,
+            })
+            .push_column_metadata(ColumnMetadata {
+                column_schema: ColumnSchema::new(
+                    "host",
+                    ConcreteDataType::string_datatype(),
+                    false,
+                ),
+                semantic_type: SemanticType::Tag,
+                column_id: 2,
+            })
+            .push_column_metadata(ColumnMetadata {
+                column_schema: ColumnSchema::new(
+                    "ts",
+                    ConcreteDataType::timestamp_millisecond_datatype(),
+                    false,
+                ),
+                semantic_type: SemanticType::Timestamp,
+                column_id: 3,
+            })
+            .primary_key(vec![1, 2]);
+        let metadata = Arc::new(builder.build().unwrap());
+
+        for (schema, expected) in [
+            (
+                full_schema.clone(),
+                Partitioning::Hash(
+                    vec![
+                        Arc::new(
+                            Column::new_with_schema("marker", full_schema.arrow_schema()).unwrap(),
+                        ) as _,
+                        Arc::new(
+                            Column::new_with_schema("host", full_schema.arrow_schema()).unwrap(),
+                        ) as _,
+                    ],
+                    2,
+                ),
+            ),
+            (
+                Arc::new(Schema::new(vec![
+                    ColumnSchema::new("marker", ConcreteDataType::int32_datatype(), false),
+                    ColumnSchema::new(
+                        "ts",
+                        ConcreteDataType::timestamp_millisecond_datatype(),
+                        false,
+                    ),
+                ])),
+                Partitioning::UnknownPartitioning(2),
+            ),
+        ] {
+            let batches = RecordBatches::try_new(schema, vec![]).unwrap();
+            let scanner = Box::new(RepeatableScanner {
+                batches,
+                properties: ScannerProperties::new(vec![vec![], vec![]], false, 0),
+                metadata: metadata.clone(),
+                dynamic_filters: Arc::new(Mutex::new(Vec::new())),
+            });
+            let request = ScanRequest {
+                distribution: Some(TimeSeriesDistribution::PerSeries),
+                ..Default::default()
+            };
+            let plan = RegionScanExec::new(scanner, request, None).unwrap();
+            let actual = plan.properties().output_partitioning();
+            match (&actual, &expected) {
+                (
+                    Partitioning::UnknownPartitioning(actual_count),
+                    Partitioning::UnknownPartitioning(expected_count),
+                ) => assert_eq!(actual_count, expected_count),
+                _ => assert_eq!(actual, &expected),
+            }
+        }
+    }
+
+    /// Returns whether the scan claims the given output ordering.
+    fn claims_ordering(plan: &RegionScanExec, names: &[&str]) -> bool {
+        plan.properties()
+            .equivalence_properties()
+            .ordering_satisfy(names.iter().map(|name| {
+                PhysicalSortExpr::new(
+                    Arc::new(Column::new_with_schema(name, &plan.arrow_schema).unwrap()),
+                    SortOptions {
+                        descending: false,
+                        nulls_first: true,
+                    },
+                )
+            }))
+            .unwrap()
+    }
+
+    /// Column schema used by the ordering fixtures below.
+    fn column_schema_of(name: &str) -> ColumnSchema {
+        let data_type = match name {
+            "ts" => ConcreteDataType::timestamp_millisecond_datatype(),
+            DATA_SCHEMA_TABLE_ID_COLUMN_NAME => ConcreteDataType::uint32_datatype(),
+            DATA_SCHEMA_TSID_COLUMN_NAME => ConcreteDataType::uint64_datatype(),
+            _ => ConcreteDataType::string_datatype(),
+        };
+        ColumnSchema::new(name, data_type, false)
+    }
+
+    /// The scan must describe its output ordering with the primary key order reported by
+    /// the metadata. Sorting the names (the removed logical-table workaround) would let a
+    /// projected column look like a row key prefix and claim an ordering the scanner does
+    /// not produce; the time index only extends a fully projected row key.
+    #[test]
+    fn test_per_series_ordering_follows_metadata_pk_order() {
+        // Deliberately not alphabetical: sorting these names would give `[a, b, z]`.
+        let mut builder = RegionMetadataBuilder::new(RegionId::new(1234, 5690));
+        for (index, name) in ["z", "a", "b", "ts"].into_iter().enumerate() {
+            builder.push_column_metadata(ColumnMetadata {
+                column_schema: column_schema_of(name),
+                semantic_type: if name == "ts" {
+                    SemanticType::Timestamp
+                } else {
+                    SemanticType::Tag
+                },
+                column_id: index as u32 + 1,
+            });
+        }
+        builder.primary_key(vec![1, 2, 3]);
+        let metadata = Arc::new(builder.build().unwrap());
+
+        // The claims must not depend on the scanner reporting a logical region.
+        for logical_region in [false, true] {
+            for (projection, claimed, unclaimed) in [
+                // Missing the first key: neither the remaining keys nor the time index are
+                // ordered.
+                (
+                    vec!["a", "b", "ts"],
+                    vec![],
+                    vec![vec!["a"], vec!["b"], vec!["ts"], vec!["a", "b"]],
+                ),
+                // Missing a key in the middle: only the prefix before it is ordered.
+                (
+                    vec!["z", "b", "ts"],
+                    vec![vec!["z"]],
+                    vec![vec!["b"], vec!["z", "b"], vec!["z", "ts"]],
+                ),
+                // A valid but incomplete prefix: the time index is not ordered after it.
+                (
+                    vec!["z", "a", "ts"],
+                    vec![vec!["z"], vec!["z", "a"]],
+                    vec![vec!["a", "z"], vec!["z", "a", "ts"]],
+                ),
+                // The whole row key followed by the time index.
+                (
+                    vec!["z", "a", "b", "ts"],
+                    vec![vec!["z", "a", "b"], vec!["z", "a", "b", "ts"]],
+                    vec![vec!["a"], vec!["a", "b", "z"]],
+                ),
+            ] {
+                let schema = Arc::new(Schema::new(
+                    projection
+                        .iter()
+                        .map(|name| column_schema_of(name))
+                        .collect(),
+                ));
+                let mut properties = ScannerProperties::new(vec![vec![], vec![]], false, 0);
+                properties.set_logical_region(logical_region);
+                let scanner = Box::new(RepeatableScanner {
+                    batches: RecordBatches::try_new(schema, vec![]).unwrap(),
+                    properties,
+                    metadata: metadata.clone(),
+                    dynamic_filters: Arc::new(Mutex::new(Vec::new())),
+                });
+                let plan = RegionScanExec::new(
+                    scanner,
+                    ScanRequest {
+                        distribution: Some(TimeSeriesDistribution::PerSeries),
+                        ..Default::default()
+                    },
+                    None,
+                )
+                .unwrap();
+
+                for ordering in claimed {
+                    assert!(
+                        claims_ordering(&plan, &ordering),
+                        "logical_region={logical_region} projection {projection:?} must claim \
+                         ordering {ordering:?}"
+                    );
+                }
+                for ordering in unclaimed {
+                    assert!(
+                        !claims_ordering(&plan, &ordering),
+                        "logical_region={logical_region} projection {projection:?} must not \
+                         claim ordering {ordering:?}"
+                    );
+                }
+            }
+        }
+    }
+
+    /// Metric TSID properties require projecting `__tsid` and restricting the scan to one
+    /// `__table_id`; hiding the physical row-key prefix must not claim a visible tag or time as
+    /// order or hash partition columns.
+    #[test]
+    fn test_metric_projection_claims_only_projected_row_key_prefix() {
+        let mut builder = RegionMetadataBuilder::new(RegionId::new(1234, 5691));
+        for (name, column_id) in [
+            (
+                DATA_SCHEMA_TABLE_ID_COLUMN_NAME,
+                ReservedColumnId::table_id(),
+            ),
+            (DATA_SCHEMA_TSID_COLUMN_NAME, ReservedColumnId::tsid()),
+            ("A", 1),
+            ("marker", 2),
+            ("ts", 3),
+        ] {
+            builder.push_column_metadata(ColumnMetadata {
+                column_schema: column_schema_of(name),
+                semantic_type: if name == "ts" {
+                    SemanticType::Timestamp
+                } else {
+                    SemanticType::Tag
+                },
+                column_id,
+            });
+        }
+        // The physical row key order: `__table_id`, `__tsid`, then the tag columns.
+        builder.primary_key(vec![
+            ReservedColumnId::table_id(),
+            ReservedColumnId::tsid(),
+            1,
+            2,
+        ]);
+        let metadata = Arc::new(builder.build().unwrap());
+
+        for (projection, partitioned_by_tsid, claimed, unclaimed) in [
+            // The logical projection drops the row key prefix, so `A` is not a prefix of the
+            // row key and neither `A` nor `ts` is ordered.
+            (
+                vec!["A", "marker", "ts"],
+                false,
+                vec![],
+                vec![vec!["A"], vec!["marker"], vec!["ts"], vec!["A", "ts"]],
+            ),
+            // Projecting internal `__tsid` plus the single-table filter preserves metric TSID properties.
+            (
+                vec![DATA_SCHEMA_TSID_COLUMN_NAME, "A", "marker", "ts"],
+                true,
+                vec![
+                    vec![DATA_SCHEMA_TSID_COLUMN_NAME],
+                    vec![DATA_SCHEMA_TSID_COLUMN_NAME, "ts"],
+                ],
+                vec![vec!["A"], vec!["ts"], vec!["A", "ts"]],
+            ),
+        ] {
+            let schema = Arc::new(Schema::new(
+                projection
+                    .iter()
+                    .map(|name| column_schema_of(name))
+                    .collect(),
+            ));
+            let mut properties = ScannerProperties::new(vec![vec![], vec![]], false, 0);
+            properties.set_logical_region(true);
+            let scanner = Box::new(RepeatableScanner {
+                batches: RecordBatches::try_new(schema, vec![]).unwrap(),
+                properties,
+                metadata: metadata.clone(),
+                dynamic_filters: Arc::new(Mutex::new(Vec::new())),
+            });
+            let plan = RegionScanExec::new(
+                scanner,
+                ScanRequest {
+                    distribution: Some(TimeSeriesDistribution::PerSeries),
+                    filters: if partitioned_by_tsid {
+                        vec![
+                            datafusion_expr::col(DATA_SCHEMA_TABLE_ID_COLUMN_NAME)
+                                .eq(datafusion_expr::lit(42u32)),
+                        ]
+                    } else {
+                        vec![]
+                    },
+                    ..Default::default()
+                },
+                None,
+            )
+            .unwrap();
+
+            let expected_partitioning = if partitioned_by_tsid {
+                Partitioning::Hash(
+                    vec![Arc::new(
+                        Column::new_with_schema(DATA_SCHEMA_TSID_COLUMN_NAME, &plan.arrow_schema)
+                            .unwrap(),
+                    ) as _],
+                    2,
+                )
+            } else {
+                Partitioning::UnknownPartitioning(2)
+            };
+            match (&plan.properties.partitioning, &expected_partitioning) {
+                (
+                    Partitioning::UnknownPartitioning(actual_count),
+                    Partitioning::UnknownPartitioning(expected_count),
+                ) => assert_eq!(actual_count, expected_count, "projection {projection:?}"),
+                (actual, expected) => assert_eq!(actual, expected, "projection {projection:?}"),
+            }
+
+            for ordering in claimed {
+                assert!(
+                    claims_ordering(&plan, &ordering),
+                    "projection {projection:?} must claim ordering {ordering:?}"
+                );
+            }
+            for ordering in unclaimed {
+                assert!(
+                    !claims_ordering(&plan, &ordering),
+                    "projection {projection:?} must not claim ordering {ordering:?}"
+                );
+            }
+        }
+    }
+
+    #[test]
+    fn test_count_statistics_require_exact_source_rows() {
+        for (append_mode, exact, expected) in [
+            (true, false, Precision::Absent),
+            (true, true, Precision::Exact(3)),
+            (false, true, Precision::Absent),
+        ] {
+            // Column `a` is nullable and `ts` is not.
+            let (schema, batch, metadata) = dynamic_filter_test_data(5685, true);
+            let scanner = Box::new(RepeatableScanner {
+                batches: RecordBatches::try_new(schema, vec![batch]).unwrap(),
+                properties: ScannerProperties::default()
+                    .with_append_mode(append_mode)
+                    .with_total_rows(3)
+                    .with_total_rows_is_exact(exact),
+                metadata,
+                dynamic_filters: Arc::new(Mutex::new(Vec::new())),
+            });
+            let plan = RegionScanExec::new(scanner, ScanRequest::default(), None).unwrap();
+            let stats = plan.partition_statistics(None).unwrap();
+            assert_eq!(stats.num_rows, expected);
+            if expected != Precision::Absent {
+                let [a, ts] = stats.column_statistics.as_slice() else {
+                    panic!("unexpected column statistics: {stats:?}");
+                };
+                assert_eq!(a.null_count, Precision::Absent);
+                assert_eq!(ts.null_count, Precision::Exact(0));
+                assert_eq!(a.distinct_count, Precision::Absent);
+                assert_eq!(ts.distinct_count, Precision::Absent);
+            }
+            assert_eq!(
+                plan.partition_statistics(Some(0)).unwrap().num_rows,
+                Precision::Absent,
+            );
+        }
+    }
+
    #[tokio::test]
    async fn test_reset_state_clears_dynamic_filters_for_repeatable_scanner() {
        let ctx = SessionContext::new();