fix(31): final gap fixes — file-backed scope filter, META-07 test, dead code removal

- file_backed_index/mod.rs: Add window_start and sort_fields filtering
  to metrics_split_matches_query() for compaction scope queries
- writer.rs: Add test_meta07_self_describing_parquet_roundtrip test
  (writes compaction metadata to Parquet, reads back from cold file,
  verifies all fields roundtrip correctly)
- fields.rs: Remove dead sort_order() method (replaced by TableConfig)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: g-talbot

quickwit/quickwit-metastore/src/metastore/file_backed/file_backed_index/mod.rs

@@ -979,6 +979,18 @@ fn metrics_split_matches_query(split: &StoredMetricsSplit, query: &ListMetricsSp
         }
     }
 
+    // Filter by compaction scope
+    if let Some(ws) = query.window_start {
+        if split.metadata.window_start != Some(ws) {
+            return false;
+        }
+    }
+    if let Some(ref sf) = query.sort_fields {
+        if split.metadata.sort_fields != *sf {
+            return false;
+        }
+    }
+
     true
 }
 

quickwit/quickwit-parquet-engine/src/schema/fields.rs

@@ -12,10 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//! Parquet field definitions with sort order constants and validation.
+//! Parquet field definitions with column metadata, sort order constants, and validation.
 
 use anyhow::{Result, bail};
-use arrow::datatypes::DataType;
+use arrow::datatypes::{DataType, Field, Fields};
+use parquet::variant::VariantType;
 
 /// Required field names that must exist in every batch.
 pub const REQUIRED_FIELDS: &[&str] = &["metric_name", "metric_type", "timestamp_secs", "value"];
@@ -31,6 +32,144 @@ pub const SORT_ORDER: &[&str] = &[
     "timestamp_secs",
 ];
 
+/// All fields in the parquet schema.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum ParquetField {
+    MetricName,
+    MetricType,
+    MetricUnit,
+    TimestampSecs,
+    StartTimestampSecs,
+    Value,
+    TagService,
+    TagEnv,
+    TagDatacenter,
+    TagRegion,
+    TagHost,
+    Attributes,
+    ServiceName,
+    ResourceAttributes,
+}
+
+impl ParquetField {
+    /// Field name as stored in Parquet.
+    pub fn name(&self) -> &'static str {
+        match self {
+            Self::MetricName => "metric_name",
+            Self::MetricType => "metric_type",
+            Self::MetricUnit => "metric_unit",
+            Self::TimestampSecs => "timestamp_secs",
+            Self::StartTimestampSecs => "start_timestamp_secs",
+            Self::Value => "value",
+            Self::TagService => "tag_service",
+            Self::TagEnv => "tag_env",
+            Self::TagDatacenter => "tag_datacenter",
+            Self::TagRegion => "tag_region",
+            Self::TagHost => "tag_host",
+            Self::Attributes => "attributes",
+            Self::ServiceName => "service_name",
+            Self::ResourceAttributes => "resource_attributes",
+        }
+    }
+
+    /// Whether this field is nullable.
+    pub fn nullable(&self) -> bool {
+        matches!(
+            self,
+            Self::MetricUnit
+                | Self::StartTimestampSecs
+                | Self::TagService
+                | Self::TagEnv
+                | Self::TagDatacenter
+                | Self::TagRegion
+                | Self::TagHost
+                | Self::Attributes
+                | Self::ResourceAttributes
+        )
+    }
+
+    /// Arrow DataType for this field.
+    /// Use dictionary encoding for high-cardinality strings.
+    pub fn arrow_type(&self) -> DataType {
+        match self {
+            // Dictionary-encoded strings for high cardinality
+            Self::MetricName | Self::ServiceName => {
+                DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8))
+            }
+            // Dictionary-encoded optional tags
+            Self::TagService
+            | Self::TagEnv
+            | Self::TagDatacenter
+            | Self::TagRegion
+            | Self::TagHost => {
+                DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8))
+            }
+            // Enum stored as UInt8
+            Self::MetricType => DataType::UInt8,
+            // Timestamps as UInt64 seconds
+            Self::TimestampSecs | Self::StartTimestampSecs => DataType::UInt64,
+            // Metric value
+            Self::Value => DataType::Float64,
+            // Plain string for metric unit
+            Self::MetricUnit => DataType::Utf8,
+            // VARIANT type for semi-structured attributes
+            // Uses the Parquet Variant binary encoding format
+            Self::Attributes | Self::ResourceAttributes => {
+                // VARIANT is stored as a struct with metadata and value BinaryView fields
+                // VariantArrayBuilder produces BinaryView, not Binary
+                DataType::Struct(Fields::from(vec![
+                    Field::new("metadata", DataType::BinaryView, false),
+                    Field::new("value", DataType::BinaryView, false),
+                ]))
+            }
+        }
+    }
+
+    /// Convert to Arrow Field.
+    pub fn to_arrow_field(&self) -> Field {
+        let field = Field::new(self.name(), self.arrow_type(), self.nullable());
+
+        // Add VARIANT extension type metadata for attributes fields
+        match self {
+            Self::Attributes | Self::ResourceAttributes => field.with_extension_type(VariantType),
+            _ => field,
+        }
+    }
+
+    /// All fields in schema order.
+    pub fn all() -> &'static [ParquetField] {
+        &[
+            Self::MetricName,
+            Self::MetricType,
+            Self::MetricUnit,
+            Self::TimestampSecs,
+            Self::StartTimestampSecs,
+            Self::Value,
+            Self::TagService,
+            Self::TagEnv,
+            Self::TagDatacenter,
+            Self::TagRegion,
+            Self::TagHost,
+            Self::Attributes,
+            Self::ServiceName,
+            Self::ResourceAttributes,
+        ]
+    }
+
+    /// Get the column index in the schema.
+    pub fn column_index(&self) -> usize {
+        Self::all().iter().position(|f| f == self).unwrap()
+    }
+
+    /// Look up a ParquetField by its Parquet column name.
+    ///
+    /// Used by the sort fields resolver to map sort schema column names
+    /// to physical schema columns.
+    pub fn from_name(name: &str) -> Option<Self> {
+        Self::all().iter().find(|f| f.name() == name).copied()
+    }
+}
+
 /// Arrow type for required fields by name.
 pub fn required_field_type(name: &str) -> Option<DataType> {
     match name {

quickwit/quickwit-parquet-engine/src/schema/mod.rs

@@ -20,5 +20,7 @@
 mod fields;
 mod parquet;
 
-pub use fields::{REQUIRED_FIELDS, SORT_ORDER, required_field_type, validate_required_fields};
+pub use fields::{
+    ParquetField, REQUIRED_FIELDS, SORT_ORDER, required_field_type, validate_required_fields,
+};
 pub use parquet::ParquetSchema;

quickwit/quickwit-parquet-engine/src/storage/writer.rs

@@ -701,6 +701,88 @@ mod tests {
         std::fs::remove_file(&path).ok();
     }
 
+    /// META-07 compliance: Prove the Parquet file is truly self-describing by
+    /// writing compaction metadata, reading it back from a cold file (no in-memory
+    /// state), and reconstructing the MetricsSplitMetadata compaction fields from
+    /// ONLY the Parquet key_value_metadata.
+    #[test]
+    fn test_meta07_self_describing_parquet_roundtrip() {
+        use std::fs::File;
+
+        use crate::split::{SplitId, TimeRange};
+        use parquet::file::reader::{FileReader, SerializedFileReader};
+
+        let sort_schema_str = "metric_name|host|env|timestamp/V2";
+        let window_start_secs: i64 = 1700006400;
+        let window_duration: u32 = 900;
+        let merge_ops: u32 = 7;
+        let row_keys_bytes: Vec<u8> = vec![0x0A, 0x03, 0x63, 0x70, 0x75];
+
+        let original = MetricsSplitMetadata::builder()
+            .split_id(SplitId::new("self-describing-test"))
+            .index_uid("metrics-prod:00000000000000000000000000")
+            .time_range(TimeRange::new(1700006400, 1700007300))
+            .window_start_secs(window_start_secs)
+            .window_duration_secs(window_duration)
+            .sort_fields(sort_schema_str)
+            .num_merge_ops(merge_ops)
+            .row_keys_proto(row_keys_bytes.clone())
+            .build();
+
+        let config = ParquetWriterConfig::default();
+        let writer = ParquetWriter::new(config, &TableConfig::default());
+        let batch = create_test_batch();
+
+        let temp_dir = std::env::temp_dir();
+        let path = temp_dir.join("test_self_describing_roundtrip.parquet");
+        writer
+            .write_to_file_with_metadata(&batch, &path, Some(&original))
+            .unwrap();
+
+        // Read phase: open a cold file and reconstruct fields from kv_metadata.
+        let file = File::open(&path).unwrap();
+        let reader = SerializedFileReader::new(file).unwrap();
+        let file_metadata = reader.metadata().file_metadata();
+        let kv_metadata = file_metadata
+            .key_value_metadata()
+            .expect("self-describing file must have kv_metadata");
+
+        let find_kv = |key: &str| -> Option<String> {
+            kv_metadata
+                .iter()
+                .find(|kv| kv.key == key)
+                .and_then(|kv| kv.value.clone())
+        };
+
+        let recovered_sort_schema = find_kv(PARQUET_META_SORT_FIELDS)
+            .expect("self-describing file must contain qh.sort_fields");
+        let recovered_window_start: i64 = find_kv(PARQUET_META_WINDOW_START)
+            .expect("self-describing file must contain qh.window_start")
+            .parse()
+            .expect("window_start must be parseable as i64");
+        let recovered_window_duration: u32 = find_kv(PARQUET_META_WINDOW_DURATION)
+            .expect("self-describing file must contain qh.window_duration_secs")
+            .parse()
+            .expect("window_duration must be parseable as u32");
+        let recovered_merge_ops: u32 = find_kv(PARQUET_META_NUM_MERGE_OPS)
+            .expect("self-describing file must contain qh.num_merge_ops")
+            .parse()
+            .expect("num_merge_ops must be parseable as u32");
+        let recovered_row_keys_b64 = find_kv(PARQUET_META_ROW_KEYS)
+            .expect("self-describing file must contain qh.row_keys");
+        let recovered_row_keys = BASE64
+            .decode(&recovered_row_keys_b64)
+            .expect("row_keys must be valid base64");
+
+        assert_eq!(recovered_sort_schema, sort_schema_str);
+        assert_eq!(recovered_window_start, window_start_secs);
+        assert_eq!(recovered_window_duration, window_duration);
+        assert_eq!(recovered_merge_ops, merge_ops);
+        assert_eq!(recovered_row_keys, row_keys_bytes);
+
+        std::fs::remove_file(&path).ok();
+    }
+
     #[test]
     fn test_build_compaction_kv_metadata_fully_populated() {
         use crate::split::{SplitId, TimeRange};