Ask the compressor plugin which encoding it produces

vortex-btrblocks/src/schemes/float/alprd.rs

@@ -44,6 +44,10 @@ impl Scheme for ALPRDScheme {
         vec![vortex_alp::ALPRD.id()]
     }
 
+    fn root_encoding(&self) -> Option<ArrayId> {
+        Some(vortex_alp::ALPRD.id())
+    }
+
     fn expected_compression_ratio(
         &self,
         data: &ArrayAndStats,

vortex-btrblocks/src/schemes/float/pco.rs

@@ -35,6 +35,10 @@ impl Scheme for PcoScheme {
         vec![vortex_pco::Pco.id()]
     }
 
+    fn root_encoding(&self) -> Option<ArrayId> {
+        Some(vortex_pco::Pco.id())
+    }
+
     fn expected_compression_ratio(
         &self,
         _data: &ArrayAndStats,

vortex-btrblocks/src/schemes/integer/pco.rs

@@ -36,6 +36,10 @@ impl Scheme for PcoScheme {
         vec![vortex_pco::Pco.id()]
     }
 
+    fn root_encoding(&self) -> Option<ArrayId> {
+        Some(vortex_pco::Pco.id())
+    }
+
     fn expected_compression_ratio(
         &self,
         data: &ArrayAndStats,

vortex-btrblocks/src/schemes/integer/runend.rs

@@ -52,6 +52,10 @@ impl Scheme for RunEndScheme {
         vec![RunEnd.id()]
     }
 
+    fn root_encoding(&self) -> Option<ArrayId> {
+        Some(RunEnd.id())
+    }
+
     /// Children: values=0, ends=1.
     fn num_children(&self) -> usize {
         2

vortex-btrblocks/src/schemes/integer/zigzag.rs

@@ -50,6 +50,10 @@ impl Scheme for ZigZagScheme {
         vec![ZigZag.id()]
     }
 
+    fn root_encoding(&self) -> Option<ArrayId> {
+        Some(ZigZag.id())
+    }
+
     /// Children: encoded=0.
     fn num_children(&self) -> usize {
         1

vortex-btrblocks/src/schemes/string/fsst.rs

@@ -54,6 +54,10 @@ impl Scheme for FSSTScheme {
         vec![FSST.id(), VarBin.id()]
     }
 
+    fn root_encoding(&self) -> Option<ArrayId> {
+        Some(FSST.id())
+    }
+
     /// Children: lengths=0, code_offsets=1.
     fn num_children(&self) -> usize {
         2

vortex-btrblocks/src/schemes/temporal.rs

@@ -59,6 +59,10 @@ impl Scheme for TemporalScheme {
         vec![DateTimeParts.id()]
     }
 
+    fn root_encoding(&self) -> Option<ArrayId> {
+        Some(DateTimeParts.id())
+    }
+
     /// Children: days=0, seconds=1, subseconds=2.
     fn num_children(&self) -> usize {
         3

vortex-btrblocks/src/tests.rs

@@ -21,6 +21,7 @@ use vortex_array::IntoArray;
 use vortex_array::VortexSessionExecute;
 use vortex_array::aggregate_fn::fns::sum::sum;
 use vortex_array::arrays::DecimalArray;
+use vortex_array::arrays::Dict;
 use vortex_array::arrays::ListArray;
 use vortex_array::arrays::PrimitiveArray;
 use vortex_array::arrays::StructArray;
@@ -50,6 +51,8 @@ use vortex_session::VortexSession;
 use crate::BtrBlocksCompressor;
 use crate::BtrBlocksCompressorBuilder;
 use crate::SESSION;
+use crate::schemes::integer::IntDictScheme;
+use crate::schemes::string::StringDictScheme;
 
 #[track_caller]
 fn assert_roundtrip(compressor: &BtrBlocksCompressor, input: &ArrayRef) -> VortexResult<ArrayRef> {
@@ -199,3 +202,56 @@ fn test_cuda_binary_zstd_follows_editions(
 
     Ok(())
 }
+
+/// `encoding_for` gives the encoding of the array that `compress` returns.
+#[rstest]
+#[case::unique_strings(
+    VarBinViewArray::from_iter_str((0..4096).map(|i| format!("{i:04} is a unique story"))).into_array(),
+)]
+#[case::few_strings(
+    VarBinViewArray::from_iter_str((0..4096).map(|i| ["red", "green", "blue"][i % 3])).into_array(),
+)]
+#[case::constant_strings(VarBinViewArray::from_iter_str(iter::repeat_n("same", 4096)).into_array())]
+#[case::null_strings(
+    VarBinViewArray::from_iter(iter::repeat_n(None::<&str>, 4096), DType::Utf8(Nullability::Nullable))
+        .into_array(),
+)]
+#[case::few_integers(PrimitiveArray::from_iter((0..4096i64).map(|i| (i % 3) * 1_000_000)).into_array())]
+#[case::unique_integers(PrimitiveArray::from_iter(0..4096i64).into_array())]
+fn test_encoding_for(#[case] input: ArrayRef) -> VortexResult<()> {
+    let compressor = BtrBlocksCompressorBuilder::from_session(&SESSION)
+        .unrestricted()
+        .build();
+    let mut ctx = SESSION.create_execution_ctx();
+
+    let encoding = compressor.0.encoding_for(&input, &mut ctx)?;
+    let compressed = compressor.compress(&input, &mut ctx)?;
+
+    assert_eq!(encoding, compressed.encoding_id());
+
+    Ok(())
+}
+
+/// A compressor that has only the dictionary schemes produces a dictionary or nothing.
+#[rstest]
+#[case::few_strings(
+    VarBinViewArray::from_iter_str((0..4096).map(|i| ["red", "green", "blue"][i % 3])).into_array(),
+    true,
+)]
+#[case::unique_strings(
+    VarBinViewArray::from_iter_str((0..4096).map(|i| format!("{i:04} is a unique story"))).into_array(),
+    false,
+)]
+fn test_dict_only_encoding_for(#[case] input: ArrayRef, #[case] is_dict: bool) -> VortexResult<()> {
+    let compressor = BtrBlocksCompressorBuilder::empty()
+        .with_new_scheme(&StringDictScheme)
+        .with_new_scheme(&IntDictScheme)
+        .build();
+    let mut ctx = SESSION.create_execution_ctx();
+
+    let encoding = compressor.0.encoding_for(&input, &mut ctx)?;
+
+    assert_eq!(encoding == vortex_array::VTable::id(&Dict), is_dict);
+
+    Ok(())
+}

vortex-compressor/src/builtins/dict/binary.rs

@@ -52,6 +52,10 @@ impl Scheme for BinaryDictScheme {
         vec![Dict.id()]
     }
 
+    fn root_encoding(&self) -> Option<ArrayId> {
+        Some(Dict.id())
+    }
+
     fn stats_options(&self) -> GenerateStatsOptions {
         GenerateStatsOptions {
             count_distinct_values: true,

vortex-compressor/src/builtins/dict/float.rs

@@ -58,6 +58,10 @@ impl Scheme for FloatDictScheme {
         vec![Dict.id()]
     }
 
+    fn root_encoding(&self) -> Option<ArrayId> {
+        Some(Dict.id())
+    }
+
     fn stats_options(&self) -> GenerateStatsOptions {
         GenerateStatsOptions {
             count_distinct_values: true,

vortex-compressor/src/builtins/dict/integer.rs

@@ -53,6 +53,10 @@ impl Scheme for IntDictScheme {
         vec![Dict.id()]
     }
 
+    fn root_encoding(&self) -> Option<ArrayId> {
+        Some(Dict.id())
+    }
+
     fn stats_options(&self) -> GenerateStatsOptions {
         GenerateStatsOptions {
             count_distinct_values: true,

vortex-compressor/src/builtins/dict/string.rs

@@ -52,6 +52,10 @@ impl Scheme for StringDictScheme {
         vec![Dict.id()]
     }
 
+    fn root_encoding(&self) -> Option<ArrayId> {
+        Some(Dict.id())
+    }
+
     fn stats_options(&self) -> GenerateStatsOptions {
         GenerateStatsOptions {
             count_distinct_values: true,