This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new fb9ab68 Support dedicated vector storage writes (#484)
fb9ab68 is described below
commit fb9ab68b53a2930e4059bc70c6fed1aab6f5509f
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Jul 8 19:00:25 2026 +0800
Support dedicated vector storage writes (#484)
---
crates/paimon/src/spec/core_options.rs | 40 +++
crates/paimon/src/spec/schema.rs | 107 +++++++
crates/paimon/src/table/data_evolution_reader.rs | 5 +-
crates/paimon/src/table/data_evolution_writer.rs | 311 +++++++++++++++++----
...e_writer.rs => dedicated_format_file_writer.rs} | 100 ++++++-
crates/paimon/src/table/mod.rs | 2 +-
crates/paimon/src/table/source.rs | 2 +-
crates/paimon/src/table/table_commit.rs | 2 +-
crates/paimon/src/table/table_scan.rs | 2 +-
crates/paimon/src/table/table_write.rs | 190 +++++++++++--
10 files changed, 664 insertions(+), 97 deletions(-)
diff --git a/crates/paimon/src/spec/core_options.rs
b/crates/paimon/src/spec/core_options.rs
index 3d5d79b..8585e37 100644
--- a/crates/paimon/src/spec/core_options.rs
+++ b/crates/paimon/src/spec/core_options.rs
@@ -47,6 +47,8 @@ const COMMIT_MAX_RETRY_WAIT_OPTION: &str =
"commit.max-retry-wait";
const FILE_COMPRESSION_OPTION: &str = "file.compression";
const FILE_COMPRESSION_ZSTD_LEVEL_OPTION: &str = "file.compression.zstd-level";
const FILE_FORMAT_OPTION: &str = "file.format";
+const VECTOR_FILE_FORMAT_OPTION: &str = "vector.file.format";
+const VECTOR_TARGET_FILE_SIZE_OPTION: &str = "vector.target-file-size";
const CHANGELOG_FILE_PREFIX_OPTION: &str = "changelog-file.prefix";
const CHANGELOG_FILE_FORMAT_OPTION: &str = "changelog-file.format";
const CHANGELOG_FILE_COMPRESSION_OPTION: &str = "changelog-file.compression";
@@ -710,6 +712,24 @@ impl<'a> CoreOptions<'a> {
.unwrap_or_else(|| self.target_file_size())
}
+ /// Dedicated vector-store file format, if configured.
+ ///
+ /// Java leaves this unset by default. When present, vector columns are
+ /// written to files named `*.vector.<format>`.
+ pub fn vector_file_format(&self) -> Option<&str> {
+ self.options
+ .get(VECTOR_FILE_FORMAT_OPTION)
+ .map(String::as_str)
+ .filter(|format| !format.trim().is_empty())
+ }
+
+ pub fn vector_target_file_size(&self) -> i64 {
+ self.options
+ .get(VECTOR_TARGET_FILE_SIZE_OPTION)
+ .and_then(|v| parse_memory_size(v))
+ .unwrap_or_else(|| self.target_file_size())
+ }
+
/// File format for data files (e.g. "parquet", "orc", "avro", "vortex").
/// Default is "parquet".
pub fn file_format(&self) -> &str {
@@ -1224,6 +1244,26 @@ mod tests {
assert_eq!(custom_core.changelog_file_stats_mode(), Some("counts"));
}
+ #[test]
+ fn test_vector_file_options_defaults_and_overrides() {
+ let default_options =
+ HashMap::from([("target-file-size".to_string(), "32
mb".to_string())]);
+ let default_core = CoreOptions::new(&default_options);
+ assert_eq!(default_core.vector_file_format(), None);
+ assert_eq!(default_core.vector_target_file_size(), 32 * 1024 * 1024);
+
+ let custom_options = HashMap::from([
+ (VECTOR_FILE_FORMAT_OPTION.to_string(), "vortex".to_string()),
+ (
+ VECTOR_TARGET_FILE_SIZE_OPTION.to_string(),
+ "64 mb".to_string(),
+ ),
+ ]);
+ let custom_core = CoreOptions::new(&custom_options);
+ assert_eq!(custom_core.vector_file_format(), Some("vortex"));
+ assert_eq!(custom_core.vector_target_file_size(), 64 * 1024 * 1024);
+ }
+
#[test]
fn test_commit_options_defaults() {
let options = HashMap::new();
diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs
index 75b35c9..cc3f17a 100644
--- a/crates/paimon/src/spec/schema.rs
+++ b/crates/paimon/src/spec/schema.rs
@@ -450,6 +450,11 @@ impl TableSchema {
&new_schema.partition_keys,
&new_schema.options,
)?;
+ Schema::validate_vector_store_fields(
+ &new_schema.fields,
+ &new_schema.partition_keys,
+ &new_schema.options,
+ )?;
PartialUpdateConfig::new(&new_schema.options)
.validate_create_mode(!new_schema.primary_keys.is_empty())?;
AggregationConfig::new(&new_schema.options)
@@ -798,6 +803,7 @@ impl Schema {
let fields = Self::normalize_fields(&fields, &partition_keys,
&primary_keys)?;
Self::validate_key_field_types(&fields, &primary_keys, &options)?;
Self::validate_blob_fields(&fields, &partition_keys, &options)?;
+ Self::validate_vector_store_fields(&fields, &partition_keys,
&options)?;
PartialUpdateConfig::new(&options).validate_create_mode(!primary_keys.is_empty())?;
AggregationConfig::new(&options).validate_create_mode(&primary_keys,
&fields)?;
Self::validate_first_row_changelog_producer(&options)?;
@@ -1049,6 +1055,52 @@ impl Schema {
Ok(())
}
+ fn validate_vector_store_fields(
+ fields: &[DataField],
+ partition_keys: &[String],
+ options: &HashMap<String, String>,
+ ) -> crate::Result<()> {
+ let vector_field_names = Self::top_level_vector_field_names(fields);
+ let core_options = CoreOptions::new(options);
+ if vector_field_names.is_empty() ||
core_options.vector_file_format().is_none() {
+ return Ok(());
+ }
+
+ if !core_options.data_evolution_enabled() {
+ return Err(crate::Error::ConfigInvalid {
+ message:
+ "Data evolution config must enabled for table with
dedicated VECTOR storage."
+ .to_string(),
+ });
+ }
+ if !core_options.row_tracking_enabled() {
+ return Err(crate::Error::ConfigInvalid {
+ message:
+ "Row tracking config must enabled for table with dedicated
VECTOR storage."
+ .to_string(),
+ });
+ }
+
+ if fields.len() == vector_field_names.len() {
+ return Err(crate::Error::ConfigInvalid {
+ message: "Table with dedicated VECTOR storage must have other
normal columns."
+ .to_string(),
+ });
+ }
+
+ let partition_key_set: HashSet<&str> =
partition_keys.iter().map(String::as_str).collect();
+ if vector_field_names
+ .iter()
+ .any(|name| partition_key_set.contains(name))
+ {
+ return Err(crate::Error::ConfigInvalid {
+ message: "The VECTOR type column can not be part of partition
keys.".to_string(),
+ });
+ }
+
+ Ok(())
+ }
+
fn validate_first_row_changelog_producer(
options: &HashMap<String, String>,
) -> crate::Result<()> {
@@ -1094,6 +1146,17 @@ impl Schema {
.collect()
}
+ /// Returns top-level Vector field names for dedicated vector-store checks.
+ fn top_level_vector_field_names(fields: &[DataField]) -> Vec<&str> {
+ fields
+ .iter()
+ .filter_map(|field| match field.data_type() {
+ DataType::Vector(_) => Some(field.name()),
+ _ => None,
+ })
+ .collect()
+ }
+
/// Returns the set of names that appear more than once.
pub fn duplicate_fields(names: &[String]) -> HashSet<String> {
let mut seen = HashMap::new();
@@ -2069,6 +2132,50 @@ mod tests {
assert_eq!(schema.fields().len(), 2);
}
+ #[test]
+ fn test_vector_store_requires_data_evolution_and_row_tracking() {
+ let err = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("embedding", vector_4f())
+ .option("vector.file.format", "vortex")
+ .build()
+ .unwrap_err();
+ assert!(
+ matches!(err, crate::Error::ConfigInvalid { ref message }
+ if message.contains("Data evolution config must enabled")),
+ "dedicated VECTOR storage should require data-evolution.enabled,
got {err:?}"
+ );
+
+ let err = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("embedding", vector_4f())
+ .option("vector.file.format", "vortex")
+ .option("data-evolution.enabled", "true")
+ .build()
+ .unwrap_err();
+ assert!(
+ matches!(err, crate::Error::ConfigInvalid { ref message }
+ if message.contains("Row tracking config must enabled")),
+ "dedicated VECTOR storage should require row-tracking.enabled, got
{err:?}"
+ );
+ }
+
+ #[test]
+ fn test_vector_store_rejects_vector_only_table() {
+ let err = Schema::builder()
+ .column("embedding", vector_4f())
+ .option("vector.file.format", "vortex")
+ .option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true")
+ .build()
+ .unwrap_err();
+ assert!(
+ matches!(err, crate::Error::ConfigInvalid { ref message }
+ if message.contains("must have other normal columns")),
+ "dedicated VECTOR storage should require a normal anchor file, got
{err:?}"
+ );
+ }
+
#[test]
fn test_apply_changes_revalidates_vector_bucket_key() {
let table_schema = TableSchema::new(
diff --git a/crates/paimon/src/table/data_evolution_reader.rs
b/crates/paimon/src/table/data_evolution_reader.rs
index 7bebb5c..ccb9cae 100644
--- a/crates/paimon/src/table/data_evolution_reader.rs
+++ b/crates/paimon/src/table/data_evolution_reader.rs
@@ -24,7 +24,7 @@ use crate::arrow::format::FilePredicates;
use crate::deletion_vector::{DeletionVector, DeletionVectorFactory};
use crate::io::FileIO;
use crate::spec::{DataField, DataFileMeta, DataType, Predicate,
ROW_ID_FIELD_NAME};
-use crate::table::blob_file_writer::is_blob_file_name;
+use crate::table::dedicated_format_file_writer::is_blob_file_name;
use crate::table::schema_manager::SchemaManager;
use crate::table::ArrowRecordBatchStream;
use crate::table::RowRange;
@@ -609,7 +609,8 @@ async fn resolve_descriptor_columns(
.downcast_ref::<arrow_array::BinaryArray>()
{
let resolved =
- super::blob_file_writer::resolve_blob_column(bin_col,
file_io).await?;
+
super::dedicated_format_file_writer::resolve_blob_column(bin_col, file_io)
+ .await?;
columns.push(Arc::new(resolved));
changed = true;
continue;
diff --git a/crates/paimon/src/table/data_evolution_writer.rs
b/crates/paimon/src/table/data_evolution_writer.rs
index 2228b78..49057e9 100644
--- a/crates/paimon/src/table/data_evolution_writer.rs
+++ b/crates/paimon/src/table/data_evolution_writer.rs
@@ -29,8 +29,8 @@
use crate::deletion_vector::{DeletionVector, DeletionVectorFactory};
use crate::io::FileIO;
use crate::spec::{
- BinaryRow, CoreOptions, DataField, DataFileMeta, DeletionVectorMeta,
FileKind, IndexFileMeta,
- IndexManifest, PartitionComputer,
+ BinaryRow, CoreOptions, DataField, DataFileMeta, DataType,
DeletionVectorMeta, FileKind,
+ IndexFileMeta, IndexManifest, PartitionComputer,
};
use crate::table::commit_message::CommitMessage;
use crate::table::data_file_writer::DataFileWriter;
@@ -48,6 +48,7 @@ use futures::TryStreamExt;
use indexmap::IndexMap;
use roaring::RoaringBitmap;
use std::collections::{HashMap, HashSet};
+use std::sync::Arc;
use uuid::Uuid;
const DELETION_VECTORS_INDEX_TYPE: &str = "DELETION_VECTORS";
@@ -884,7 +885,9 @@ struct MatchedRow {
// ---------------------------------------------------------------------------
/// Key: (partition_bytes, bucket, first_row_id)
-type WriterKey = (Vec<u8>, i32, i64);
+type WriterBaseKey = (Vec<u8>, i32, i64);
+/// Key: (partition_bytes, bucket, first_row_id, file kind)
+type WriterKey = (Vec<u8>, i32, i64, PartialFileKind);
type PartialCommitGroup = (Option<i64>, Vec<DataFileMeta>);
/// Writer for data evolution partial-column files.
@@ -894,25 +897,41 @@ type PartialCommitGroup = (Option<i64>,
Vec<DataFileMeta>);
/// Each output file contains only the updated columns and shares the same
`first_row_id` range
/// as the original file, allowing the reader to merge columns at read time.
///
-/// Produces parquet files containing only the specified `write_columns`, with
+/// Produces files containing only the specified `write_columns`, with
/// `file_source = APPEND (0)`, caller-supplied `first_row_id`, and
`write_cols`.
+/// VECTOR columns use dedicated `*.vector.<format>` files when
`vector.file.format`
+/// is configured.
pub(crate) struct DataEvolutionPartialWriter {
file_io: FileIO,
table_location: String,
partition_computer: PartitionComputer,
partition_keys: Vec<String>,
schema_id: i64,
- target_file_size: i64,
file_compression: String,
file_compression_zstd_level: i32,
write_buffer_size: i64,
- file_format: String,
format_options: HashMap<String, String>,
+ write_sets: Vec<PartialWriteSet>,
+ /// Writers keyed by (partition_bytes, bucket, first_row_id, file kind).
+ writers: HashMap<WriterKey, DataFileWriter>,
+ check_from_snapshots: HashMap<WriterBaseKey, i64>,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+enum PartialFileKind {
+ Normal,
+ Vector,
+}
+
+#[derive(Clone)]
+struct PartialWriteSet {
+ kind: PartialFileKind,
+ target_file_size: i64,
+ file_format: String,
write_fields: Vec<DataField>,
write_columns: Vec<String>,
- /// Writers keyed by (partition_bytes, bucket, first_row_id).
- writers: HashMap<WriterKey, DataFileWriter>,
- check_from_snapshots: HashMap<WriterKey, i64>,
+ column_indices: Vec<usize>,
+ schema: Arc<arrow_schema::Schema>,
}
impl DataEvolutionPartialWriter {
@@ -932,19 +951,7 @@ impl DataEvolutionPartialWriter {
let partition_keys: Vec<String> = schema.partition_keys().to_vec();
let fields = schema.fields();
- let write_fields = write_columns
- .iter()
- .map(|column| {
- fields
- .iter()
- .find(|field| field.name() == column)
- .cloned()
- .ok_or_else(|| crate::Error::DataInvalid {
- message: format!("Unknown data-evolution write column
'{column}'"),
- source: None,
- })
- })
- .collect::<Result<Vec<_>>>()?;
+ let write_sets = Self::build_write_sets(&write_columns, fields,
&core_options)?;
let partition_computer = PartitionComputer::new(
&partition_keys,
fields,
@@ -958,19 +965,83 @@ impl DataEvolutionPartialWriter {
partition_computer,
partition_keys,
schema_id: schema.id(),
- target_file_size: core_options.target_file_size(),
file_compression: core_options.file_compression().to_string(),
file_compression_zstd_level:
core_options.file_compression_zstd_level(),
write_buffer_size: core_options.write_parquet_buffer_size(),
- file_format: core_options.file_format().to_string(),
format_options: schema.options().clone(),
- write_fields,
- write_columns,
+ write_sets,
writers: HashMap::new(),
check_from_snapshots: HashMap::new(),
})
}
+ fn build_write_sets(
+ write_columns: &[String],
+ fields: &[DataField],
+ core_options: &CoreOptions<'_>,
+ ) -> Result<Vec<PartialWriteSet>> {
+ let vector_file_format = core_options
+ .vector_file_format()
+ .map(|format| format.trim().to_ascii_lowercase());
+ let mut normal_fields = Vec::new();
+ let mut normal_columns = Vec::new();
+ let mut normal_indices = Vec::new();
+ let mut vector_fields = Vec::new();
+ let mut vector_columns = Vec::new();
+ let mut vector_indices = Vec::new();
+
+ for (idx, column) in write_columns.iter().enumerate() {
+ let field = fields
+ .iter()
+ .find(|field| field.name() == column)
+ .cloned()
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: format!("Unknown data-evolution write column
'{column}'"),
+ source: None,
+ })?;
+
+ if vector_file_format.is_some() && matches!(field.data_type(),
DataType::Vector(_)) {
+ vector_fields.push(field);
+ vector_columns.push(column.clone());
+ vector_indices.push(idx);
+ } else {
+ normal_fields.push(field);
+ normal_columns.push(column.clone());
+ normal_indices.push(idx);
+ }
+ }
+
+ let mut write_sets = Vec::new();
+ if !normal_fields.is_empty() {
+ let schema =
crate::arrow::build_target_arrow_schema(&normal_fields)?;
+ write_sets.push(PartialWriteSet {
+ kind: PartialFileKind::Normal,
+ target_file_size: core_options.target_file_size(),
+ file_format: core_options.file_format().to_string(),
+ write_fields: normal_fields,
+ write_columns: normal_columns,
+ column_indices: normal_indices,
+ schema,
+ });
+ }
+
+ if !vector_fields.is_empty() {
+ let schema =
crate::arrow::build_target_arrow_schema(&vector_fields)?;
+ let vector_file_format = vector_file_format.expect("vector fields
require a format");
+ write_sets.push(PartialWriteSet {
+ kind: PartialFileKind::Vector,
+ target_file_size: core_options.vector_target_file_size(),
+ file_format: format!("vector.{vector_file_format}"),
+ write_fields: vector_fields,
+ write_columns: vector_columns,
+ column_indices: vector_indices,
+ schema,
+ });
+ }
+
+ Ok(write_sets)
+ }
+
/// Write a partial-column batch for a specific partition, bucket, and row
ID range.
///
/// The `batch` must contain only the columns specified in `write_columns`.
@@ -987,55 +1058,85 @@ impl DataEvolutionPartialWriter {
return Ok(());
}
- let key = (partition_bytes.clone(), bucket, first_row_id);
+ let base_key = (partition_bytes.clone(), bucket, first_row_id);
self.check_from_snapshots
- .entry(key.clone())
+ .entry(base_key.clone())
.and_modify(|snapshot| *snapshot =
(*snapshot).min(check_from_snapshot))
.or_insert(check_from_snapshot);
- if !self.writers.contains_key(&key) {
- let partition_path = if self.partition_keys.is_empty() {
- String::new()
- } else {
- let row = BinaryRow::from_serialized_bytes(&partition_bytes)?;
- self.partition_computer.generate_partition_path(&row)?
- };
- let writer = DataFileWriter::new(
- self.file_io.clone(),
- self.table_location.clone(),
- partition_path,
+ let partition_path = if self.partition_keys.is_empty() {
+ String::new()
+ } else {
+ let row = BinaryRow::from_serialized_bytes(&partition_bytes)?;
+ self.partition_computer.generate_partition_path(&row)?
+ };
+
+ for write_set in self.write_sets.clone() {
+ let key = (
+ partition_bytes.clone(),
bucket,
- self.schema_id,
- self.target_file_size,
- self.file_compression.clone(),
- self.file_compression_zstd_level,
- self.write_buffer_size,
- self.file_format.clone(),
- self.write_fields.clone(),
- self.format_options.clone(),
- Some(0), // file_source: APPEND
- Some(first_row_id),
- Some(self.write_columns.clone()),
+ first_row_id,
+ write_set.kind,
);
- self.writers.insert(key.clone(), writer);
+ if !self.writers.contains_key(&key) {
+ let writer = DataFileWriter::new(
+ self.file_io.clone(),
+ self.table_location.clone(),
+ partition_path.clone(),
+ bucket,
+ self.schema_id,
+ write_set.target_file_size,
+ self.file_compression.clone(),
+ self.file_compression_zstd_level,
+ self.write_buffer_size,
+ write_set.file_format.clone(),
+ write_set.write_fields.clone(),
+ self.format_options.clone(),
+ Some(0), // file_source: APPEND
+ Some(first_row_id),
+ Some(write_set.write_columns.clone()),
+ );
+ self.writers.insert(key.clone(), writer);
+ }
+
+ let projected_batch = Self::project_batch(&batch, &write_set)?;
+ let writer = self.writers.get_mut(&key).unwrap();
+ writer.write(&projected_batch).await?;
}
- let writer = self.writers.get_mut(&key).unwrap();
- writer.write(&batch).await
+ Ok(())
+ }
+
+ fn project_batch(batch: &RecordBatch, write_set: &PartialWriteSet) ->
Result<RecordBatch> {
+ let columns: Vec<ArrayRef> = write_set
+ .column_indices
+ .iter()
+ .map(|&idx| batch.column(idx).clone())
+ .collect();
+ RecordBatch::try_new(write_set.schema.clone(), columns).map_err(|e| {
+ crate::Error::DataInvalid {
+ message: format!(
+ "Failed to project data-evolution columns {:?}: {e}",
+ write_set.write_columns
+ ),
+ source: None,
+ }
+ })
}
/// Close all writers and collect CommitMessages for use with TableCommit.
pub async fn prepare_commit(&mut self) -> Result<Vec<CommitMessage>> {
let writers: Vec<(WriterKey, DataFileWriter)> =
self.writers.drain().collect();
- let mut check_from_snapshots = std::mem::take(&mut
self.check_from_snapshots);
+ let check_from_snapshots = std::mem::take(&mut
self.check_from_snapshots);
let futures: Vec<_> = writers
.into_iter()
.map(|(key, mut writer)| {
- let check_from_snapshot = check_from_snapshots.remove(&key);
+ let base_key = (key.0.clone(), key.1, key.2);
+ let check_from_snapshot =
check_from_snapshots.get(&base_key).copied();
async move {
let files = writer.prepare_commit().await?;
- let (partition_bytes, bucket, _first_row_id) = key;
+ let (partition_bytes, bucket, _first_row_id, _kind) = key;
Ok::<_, crate::Error>((partition_bytes, bucket,
check_from_snapshot, files))
}
})
@@ -1074,7 +1175,7 @@ mod tests {
use super::*;
use crate::catalog::Identifier;
use crate::io::FileIOBuilder;
- use crate::spec::{DataType, IntType, Schema, TableSchema, VarCharType};
+ use crate::spec::{DataType, FloatType, IntType, Schema, TableSchema,
VarCharType, VectorType};
use arrow_array::StringArray;
use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema
as ArrowSchema};
use std::sync::Arc;
@@ -1128,6 +1229,22 @@ mod tests {
TableSchema::new(0, &schema)
}
+ fn test_vector_data_evolution_schema(vector_file_format: &str) ->
TableSchema {
+ let vector_type = DataType::Vector(
+ VectorType::try_new(true, 2,
DataType::Float(FloatType::new())).unwrap(),
+ );
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("name", DataType::VarChar(VarCharType::string_type()))
+ .column("embedding", vector_type)
+ .option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true")
+ .option("vector.file.format", vector_file_format)
+ .build()
+ .unwrap();
+ TableSchema::new(0, &schema)
+ }
+
fn test_table(file_io: &FileIO, table_path: &str) -> Table {
Table::new(
file_io.clone(),
@@ -1158,6 +1275,36 @@ mod tests {
RecordBatch::try_new(schema,
vec![Arc::new(StringArray::from(names))]).unwrap()
}
+ fn make_partial_name_vector_batch(names: Vec<&str>, vectors:
Vec<Vec<f32>>) -> RecordBatch {
+ use arrow_array::builder::{FixedSizeListBuilder, Float32Builder};
+
+ let element_field = Arc::new(ArrowField::new("element",
ArrowDataType::Float32, true));
+ let mut builder =
+ FixedSizeListBuilder::new(Float32Builder::new(),
2).with_field(element_field.clone());
+ for vector in vectors {
+ assert_eq!(vector.len(), 2);
+ for value in vector {
+ builder.values().append_value(value);
+ }
+ builder.append(true);
+ }
+ let embedding = builder.finish();
+
+ let schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("name", ArrowDataType::Utf8, true),
+ ArrowField::new(
+ "embedding",
+ ArrowDataType::FixedSizeList(element_field, 2),
+ true,
+ ),
+ ]));
+ RecordBatch::try_new(
+ schema,
+ vec![Arc::new(StringArray::from(names)), Arc::new(embedding)],
+ )
+ .unwrap()
+ }
+
fn make_matched_batch(row_ids: Vec<Option<i64>>, names: Vec<&str>) ->
RecordBatch {
let schema = Arc::new(ArrowSchema::new(vec![
ArrowField::new("_ROW_ID", ArrowDataType::Int64, true),
@@ -1221,6 +1368,56 @@ mod tests {
assert_eq!(meta.file_source, Some(0));
}
+ #[tokio::test]
+ async fn test_write_partial_vector_column_uses_dedicated_file() {
+ let file_io = test_file_io();
+ let table_path = "memory:/test_de_write_vector";
+ setup_dirs(&file_io, table_path).await;
+
+ let table = Table::new(
+ file_io,
+ Identifier::new("default", "test_de_vector_table"),
+ table_path.to_string(),
+ test_vector_data_evolution_schema("parquet"),
+ None,
+ );
+ let mut writer = DataEvolutionPartialWriter::new(
+ &table,
+ vec!["name".to_string(), "embedding".to_string()],
+ )
+ .unwrap();
+
+ let batch = make_partial_name_vector_batch(
+ vec!["alice", "bob"],
+ vec![vec![1.0, 0.0], vec![0.0, 1.0]],
+ );
+ writer
+ .write_partial_batch(vec![], 0, 42, 7, batch)
+ .await
+ .unwrap();
+
+ let messages = writer.prepare_commit().await.unwrap();
+ assert_eq!(messages.len(), 1);
+ assert_eq!(messages[0].new_files.len(), 2);
+ assert_eq!(messages[0].check_from_snapshot, Some(7));
+
+ let normal_file = messages[0]
+ .new_files
+ .iter()
+ .find(|file| !file.file_name.contains(".vector."))
+ .unwrap();
+ let vector_file = messages[0]
+ .new_files
+ .iter()
+ .find(|file| file.file_name.ends_with(".vector.parquet"))
+ .unwrap();
+
+ assert_eq!(normal_file.first_row_id, Some(42));
+ assert_eq!(normal_file.write_cols, Some(vec!["name".to_string()]));
+ assert_eq!(vector_file.first_row_id, Some(42));
+ assert_eq!(vector_file.write_cols,
Some(vec!["embedding".to_string()]));
+ }
+
#[tokio::test]
async fn test_different_first_row_id_creates_separate_files() {
let file_io = test_file_io();
diff --git a/crates/paimon/src/table/blob_file_writer.rs
b/crates/paimon/src/table/dedicated_format_file_writer.rs
similarity index 66%
rename from crates/paimon/src/table/blob_file_writer.rs
rename to crates/paimon/src/table/dedicated_format_file_writer.rs
index b6572f6..347d435 100644
--- a/crates/paimon/src/table/blob_file_writer.rs
+++ b/crates/paimon/src/table/dedicated_format_file_writer.rs
@@ -16,7 +16,7 @@
// under the License.
use crate::io::FileIO;
-use crate::spec::{BlobDescriptor, DataFileMeta};
+use crate::spec::{BlobDescriptor, DataField, DataFileMeta, DataType};
use crate::table::data_file_writer::DataFileWriter;
use crate::Result;
use arrow_array::builder::BinaryBuilder;
@@ -34,22 +34,32 @@ struct BlobFieldWriter {
column_index: usize,
}
-/// Writes append-only data with blob columns split into separate `.blob`
files.
+struct VectorFieldWriter {
+ writer: DataFileWriter,
+ field_names: Vec<String>,
+ column_indices: Vec<usize>,
+ schema: Arc<arrow_schema::Schema>,
+}
+
+/// Writes append-only data with columns split into dedicated file formats.
///
-/// Normal (non-blob) columns go to a parquet `DataFileWriter`.
-/// Each blob column (not in `blob_descriptor_fields`) gets its own
`DataFileWriter`
-/// with `file_format = "blob"` and `write_cols = Some(vec![field_name])`.
+/// Remaining columns go to the table's normal append `DataFileWriter`.
+/// Each non-descriptor blob column gets its own `DataFileWriter` with
+/// `file_format = "blob"` and `write_cols = Some(vec![field_name])`.
+/// When `vector.file.format` is configured, all VECTOR columns are written
+/// together to a dedicated `*.vector.<format>` file.
///
/// If a blob value is already a serialized `BlobDescriptor`, the actual data
is
/// resolved from the referenced URI and written to the `.blob` file.
-pub(crate) struct AppendBlobFileWriter {
+pub(crate) struct AppendDedicatedFormatFileWriter {
normal_writer: DataFileWriter,
blob_writers: Vec<BlobFieldWriter>,
+ vector_writer: Option<VectorFieldWriter>,
normal_column_indices: Vec<usize>,
normal_schema: Arc<arrow_schema::Schema>,
}
-impl AppendBlobFileWriter {
+impl AppendDedicatedFormatFileWriter {
#[allow(clippy::too_many_arguments)]
pub(crate) fn new(
file_io: FileIO,
@@ -63,8 +73,10 @@ impl AppendBlobFileWriter {
file_compression_zstd_level: i32,
write_buffer_size: i64,
file_format: String,
+ vector_target_file_size: i64,
+ vector_file_format: Option<&str>,
input_schema: &arrow_schema::Schema,
- table_fields: &[crate::spec::DataField],
+ table_fields: &[DataField],
format_options: &HashMap<String, String>,
blob_descriptor_fields: &HashSet<String>,
) -> Self {
@@ -72,12 +84,23 @@ impl AppendBlobFileWriter {
let mut normal_arrow_fields = Vec::new();
let mut normal_table_fields = Vec::new();
let mut blob_writers = Vec::new();
+ let mut vector_column_indices = Vec::new();
+ let mut vector_arrow_fields = Vec::new();
+ let mut vector_table_fields = Vec::new();
+ let mut vector_field_names = Vec::new();
for (idx, field) in table_fields.iter().enumerate() {
let is_blob = field.data_type().is_blob_type();
let is_descriptor = blob_descriptor_fields.contains(field.name());
+ let is_dedicated_vector =
+ vector_file_format.is_some() && matches!(field.data_type(),
DataType::Vector(_));
- if is_blob && !is_descriptor {
+ if is_dedicated_vector {
+ vector_column_indices.push(idx);
+ vector_arrow_fields.push(input_schema.field(idx).clone());
+ vector_table_fields.push(field.clone());
+ vector_field_names.push(field.name().to_string());
+ } else if is_blob && !is_descriptor {
blob_writers.push(BlobFieldWriter {
writer: DataFileWriter::new(
file_io.clone(),
@@ -107,6 +130,37 @@ impl AppendBlobFileWriter {
}
let normal_schema =
Arc::new(arrow_schema::Schema::new(normal_arrow_fields));
+ let vector_writer = if let Some(vector_file_format) =
vector_file_format {
+ if vector_table_fields.is_empty() {
+ None
+ } else {
+ let vector_schema =
Arc::new(arrow_schema::Schema::new(vector_arrow_fields));
+ Some(VectorFieldWriter {
+ writer: DataFileWriter::new(
+ file_io.clone(),
+ table_location.clone(),
+ partition_path.clone(),
+ bucket,
+ schema_id,
+ vector_target_file_size,
+ file_compression.clone(),
+ file_compression_zstd_level,
+ write_buffer_size,
+ format!("vector.{}",
vector_file_format.trim().to_ascii_lowercase()),
+ vector_table_fields,
+ format_options.clone(),
+ Some(0),
+ None,
+ Some(vector_field_names.clone()),
+ ),
+ field_names: vector_field_names,
+ column_indices: vector_column_indices,
+ schema: vector_schema,
+ })
+ }
+ } else {
+ None
+ };
let normal_writer = DataFileWriter::new(
file_io.clone(),
@@ -129,6 +183,7 @@ impl AppendBlobFileWriter {
Self {
normal_writer,
blob_writers,
+ vector_writer,
normal_column_indices,
normal_schema,
}
@@ -170,6 +225,28 @@ impl AppendBlobFileWriter {
blob_writer.writer.write(&blob_batch).await?;
}
+ if let Some(vector_writer) = &mut self.vector_writer {
+ let vector_columns: Vec<Arc<dyn arrow_array::Array>> =
vector_writer
+ .column_indices
+ .iter()
+ .map(|&idx| batch.column(idx).clone())
+ .collect();
+ let vector_batch =
+ match RecordBatch::try_new(vector_writer.schema.clone(),
vector_columns) {
+ Ok(batch) => batch,
+ Err(e) => {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Failed to project vector columns {:?}: {e}",
+ vector_writer.field_names
+ ),
+ source: None,
+ });
+ }
+ };
+ vector_writer.writer.write(&vector_batch).await?;
+ }
+
Ok(())
}
@@ -181,6 +258,11 @@ impl AppendBlobFileWriter {
results.extend(blob_metas);
}
+ if let Some(vector_writer) = &mut self.vector_writer {
+ let vector_metas = vector_writer.writer.prepare_commit().await?;
+ results.extend(vector_metas);
+ }
+
Ok(results)
}
}
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index 70eef2d..28e3a6a 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -20,7 +20,6 @@
pub(crate) mod aggregator;
pub(crate) mod bin_pack;
mod bitmap_global_index_reader;
-mod blob_file_writer;
mod branch_manager;
mod btree_global_index_build_builder;
mod btree_global_index_drop_builder;
@@ -37,6 +36,7 @@ mod data_evolution_reader;
pub mod data_evolution_writer;
mod data_file_reader;
mod data_file_writer;
+mod dedicated_format_file_writer;
#[cfg(feature = "fulltext")]
mod full_text_search_builder;
pub(crate) mod global_index_build_common;
diff --git a/crates/paimon/src/table/source.rs
b/crates/paimon/src/table/source.rs
index 88ccfd0..c4254b6 100644
--- a/crates/paimon/src/table/source.rs
+++ b/crates/paimon/src/table/source.rs
@@ -28,7 +28,7 @@ fn is_vector_store_file_name(file_name: &str) -> bool {
}
pub(crate) fn is_data_evolution_normal_file(file: &DataFileMeta) -> bool {
- !crate::table::blob_file_writer::is_blob_file_name(&file.file_name)
+
!crate::table::dedicated_format_file_writer::is_blob_file_name(&file.file_name)
&& !is_vector_store_file_name(&file.file_name)
}
diff --git a/crates/paimon/src/table/table_commit.rs
b/crates/paimon/src/table/table_commit.rs
index 453d8aa..76aaf50 100644
--- a/crates/paimon/src/table/table_commit.rs
+++ b/crates/paimon/src/table/table_commit.rs
@@ -2622,7 +2622,7 @@ fn ranges_overlap(left_start: i64, left_end: i64,
right_start: i64, right_end: i
}
fn is_blob_data_file(file: &DataFileMeta) -> bool {
- crate::table::blob_file_writer::is_blob_file_name(&file.file_name)
+
crate::table::dedicated_format_file_writer::is_blob_file_name(&file.file_name)
}
fn is_vector_store_file(file: &DataFileMeta) -> bool {
diff --git a/crates/paimon/src/table/table_scan.rs
b/crates/paimon/src/table/table_scan.rs
index 67184a2..71ad467 100644
--- a/crates/paimon/src/table/table_scan.rs
+++ b/crates/paimon/src/table/table_scan.rs
@@ -455,7 +455,7 @@ fn is_vector_store_file_name(file_name: &str) -> bool {
}
fn is_normal_data_file(file: &DataFileMeta) -> bool {
- !crate::table::blob_file_writer::is_blob_file_name(&file.file_name)
+
!crate::table::dedicated_format_file_writer::is_blob_file_name(&file.file_name)
&& !is_vector_store_file_name(&file.file_name)
}
diff --git a/crates/paimon/src/table/table_write.rs
b/crates/paimon/src/table/table_write.rs
index 157d3cb..1854c53 100644
--- a/crates/paimon/src/table/table_write.rs
+++ b/crates/paimon/src/table/table_write.rs
@@ -26,7 +26,6 @@ use crate::spec::{
first_row_supports_changelog_producer, BinaryRow, ChangelogProducer,
CoreOptions, DataField,
DataType, MergeEngine, EMPTY_SERIALIZED_ROW, POSTPONE_BUCKET,
};
-use crate::table::blob_file_writer::AppendBlobFileWriter;
use crate::table::bucket_assigner::{BucketAssignerEnum, PartitionBucketKey};
use crate::table::bucket_assigner_constant::ConstantBucketAssigner;
use crate::table::bucket_assigner_cross::CrossPartitionAssigner;
@@ -35,6 +34,7 @@ use crate::table::bucket_assigner_fixed::FixedBucketAssigner;
use crate::table::bucket_function::validate_bucket_function;
use crate::table::commit_message::CommitMessage;
use crate::table::data_file_writer::DataFileWriter;
+use
crate::table::dedicated_format_file_writer::AppendDedicatedFormatFileWriter;
use crate::table::kv_file_writer::{KeyValueFileWriter, KeyValueWriteConfig};
use crate::table::partition_filter::PartitionFilter;
use crate::table::postpone_file_writer::{PostponeFileWriter,
PostponeWriteConfig};
@@ -48,7 +48,7 @@ use std::sync::Arc;
/// Enum to hold either an append-only writer, a key-value writer, or a
postpone writer.
enum FileWriter {
Append(DataFileWriter),
- AppendBlob(AppendBlobFileWriter),
+ AppendDedicated(Box<AppendDedicatedFormatFileWriter>),
KeyValue(KeyValueFileWriter),
Postpone(PostponeFileWriter),
}
@@ -57,7 +57,7 @@ impl FileWriter {
async fn write(&mut self, batch: &RecordBatch) -> Result<()> {
match self {
FileWriter::Append(w) => w.write(batch).await,
- FileWriter::AppendBlob(w) => w.write(batch).await,
+ FileWriter::AppendDedicated(w) => w.write(batch).await,
FileWriter::KeyValue(w) => w.write(batch).await,
FileWriter::Postpone(w) => w.write(batch).await,
}
@@ -66,7 +66,9 @@ impl FileWriter {
async fn prepare_commit(mut self) -> Result<PreparedFiles> {
match self {
FileWriter::Append(ref mut w) =>
w.prepare_commit().await.map(PreparedFiles::data),
- FileWriter::AppendBlob(ref mut w) =>
w.prepare_commit().await.map(PreparedFiles::data),
+ FileWriter::AppendDedicated(ref mut w) => {
+ w.prepare_commit().await.map(PreparedFiles::data)
+ }
FileWriter::KeyValue(ref mut w) => w.prepare_commit().await,
FileWriter::Postpone(ref mut w) =>
w.prepare_commit().await.map(PreparedFiles::data),
}
@@ -90,6 +92,7 @@ pub struct TableWrite {
schema_id: i64,
target_file_size: i64,
blob_target_file_size: i64,
+ vector_target_file_size: i64,
file_compression: String,
file_compression_zstd_level: i32,
write_buffer_size: i64,
@@ -110,8 +113,12 @@ pub struct TableWrite {
is_overwrite: bool,
/// Blob descriptor fields (stored inline in parquet, not as separate
.blob files).
blob_descriptor_fields: HashSet<String>,
- /// Whether the table has non-descriptor blob fields requiring
AppendBlobFileWriter.
+ /// Whether the table has non-descriptor blob fields requiring a
dedicated-format writer.
has_blob_fields: bool,
+ /// Dedicated vector-store file format, when configured.
+ vector_file_format: Option<String>,
+ /// Whether the table has VECTOR fields requiring dedicated vector files.
+ has_dedicated_vector_fields: bool,
}
impl TableWrite {
@@ -175,9 +182,11 @@ impl TableWrite {
}
let target_file_size = core_options.target_file_size();
let blob_target_file_size = core_options.blob_target_file_size();
+ let vector_target_file_size = core_options.vector_target_file_size();
let file_compression = core_options.file_compression().to_string();
let file_compression_zstd_level =
core_options.file_compression_zstd_level();
let file_format = core_options.file_format().to_string();
+ let vector_file_format =
core_options.vector_file_format().map(str::to_string);
let changelog_file_prefix =
core_options.changelog_file_prefix().to_string();
let changelog_file_format =
core_options.changelog_file_format().to_string();
let changelog_file_compression =
core_options.changelog_file_compression().to_string();
@@ -305,6 +314,11 @@ impl TableWrite {
.fields()
.iter()
.any(|f| f.data_type().is_blob_type() &&
!blob_descriptor_fields.contains(f.name()));
+ let has_dedicated_vector_fields = vector_file_format.is_some()
+ && schema
+ .fields()
+ .iter()
+ .any(|f| matches!(f.data_type(), DataType::Vector(_)));
Ok(Self {
table: table.clone(),
@@ -314,6 +328,7 @@ impl TableWrite {
schema_id: schema.id(),
target_file_size,
blob_target_file_size,
+ vector_target_file_size,
file_compression,
file_compression_zstd_level,
write_buffer_size,
@@ -332,6 +347,8 @@ impl TableWrite {
is_overwrite,
blob_descriptor_fields,
has_blob_fields,
+ vector_file_format,
+ has_dedicated_vector_fields,
})
}
@@ -624,25 +641,29 @@ impl TableWrite {
/// Create an append-only writer for non-PK tables.
fn create_append_writer(&self, partition_path: String, bucket: i32) ->
Result<FileWriter> {
- if self.has_blob_fields {
+ if self.has_blob_fields || self.has_dedicated_vector_fields {
let fields = self.table.schema().fields();
let input_schema = build_target_arrow_schema(fields)?;
- Ok(FileWriter::AppendBlob(AppendBlobFileWriter::new(
- self.table.file_io().clone(),
- self.table.location().to_string(),
- partition_path,
- bucket,
- self.schema_id,
- self.target_file_size,
- self.blob_target_file_size,
- self.file_compression.clone(),
- self.file_compression_zstd_level,
- self.write_buffer_size,
- self.file_format.clone(),
- &input_schema,
- fields,
- self.table.schema().options(),
- &self.blob_descriptor_fields,
+ Ok(FileWriter::AppendDedicated(Box::new(
+ AppendDedicatedFormatFileWriter::new(
+ self.table.file_io().clone(),
+ self.table.location().to_string(),
+ partition_path,
+ bucket,
+ self.schema_id,
+ self.target_file_size,
+ self.blob_target_file_size,
+ self.file_compression.clone(),
+ self.file_compression_zstd_level,
+ self.write_buffer_size,
+ self.file_format.clone(),
+ self.vector_target_file_size,
+ self.vector_file_format.as_deref(),
+ &input_schema,
+ fields,
+ self.table.schema().options(),
+ &self.blob_descriptor_fields,
+ ),
)))
} else {
Ok(FileWriter::Append(DataFileWriter::new(
@@ -745,9 +766,10 @@ mod tests {
use crate::io::{FileIO, FileIOBuilder};
use crate::spec::{
bucket_dir_name, BigIntType, BinaryRowBuilder, BlobType, DataField,
DataType, DecimalType,
- FileKind, IndexManifest, IntType, LocalZonedTimestampType, Manifest,
ManifestList, Schema,
- TableSchema, TimestampType, TinyIntType, VarCharType,
SEQUENCE_NUMBER_FIELD_ID,
- SEQUENCE_NUMBER_FIELD_NAME, VALUE_KIND_FIELD_ID, VALUE_KIND_FIELD_NAME,
+ FileKind, FloatType, IndexManifest, IntType, LocalZonedTimestampType,
Manifest,
+ ManifestList, Schema, TableSchema, TimestampType, TinyIntType,
VarCharType, VectorType,
+ SEQUENCE_NUMBER_FIELD_ID, SEQUENCE_NUMBER_FIELD_NAME,
VALUE_KIND_FIELD_ID,
+ VALUE_KIND_FIELD_NAME,
};
use crate::table::{SnapshotManager, TableCommit};
use arrow_array::RecordBatchReader as _;
@@ -810,6 +832,21 @@ mod tests {
TableSchema::new(0, &schema)
}
+ fn test_vector_table_schema(vector_file_format: &str) -> TableSchema {
+ let vector_type = DataType::Vector(
+ VectorType::try_new(true, 2,
DataType::Float(FloatType::new())).unwrap(),
+ );
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("embedding", vector_type)
+ .option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true")
+ .option("vector.file.format", vector_file_format)
+ .build()
+ .unwrap();
+ TableSchema::new(0, &schema)
+ }
+
async fn setup_dirs(file_io: &FileIO, table_path: &str) {
file_io
.mkdirs(&format!("{table_path}/snapshot/"))
@@ -836,6 +873,36 @@ mod tests {
.unwrap()
}
+ fn make_vector_batch(ids: Vec<i32>, vectors: Vec<Vec<f32>>) -> RecordBatch
{
+ use arrow_array::builder::{FixedSizeListBuilder, Float32Builder};
+
+ let element_field = Arc::new(ArrowField::new("element",
ArrowDataType::Float32, true));
+ let mut builder =
+ FixedSizeListBuilder::new(Float32Builder::new(),
2).with_field(element_field.clone());
+ for vector in vectors {
+ assert_eq!(vector.len(), 2);
+ for value in vector {
+ builder.values().append_value(value);
+ }
+ builder.append(true);
+ }
+ let embedding = builder.finish();
+
+ let schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new(
+ "embedding",
+ ArrowDataType::FixedSizeList(element_field, 2),
+ true,
+ ),
+ ]));
+ RecordBatch::try_new(
+ schema,
+ vec![Arc::new(Int32Array::from(ids)), Arc::new(embedding)],
+ )
+ .unwrap()
+ }
+
fn make_batch_with_value_kind(
ids: Vec<i32>,
values: Vec<i32>,
@@ -1156,6 +1223,79 @@ mod tests {
assert_eq!(snapshot.id(), 1);
}
+ async fn assert_vector_write_uses_dedicated_file(table_path: &str,
vector_file_format: &str) {
+ let file_io = test_file_io();
+ setup_dirs(&file_io, table_path).await;
+
+ let table = Table::new(
+ file_io.clone(),
+ Identifier::new("default", "test_vector_table"),
+ table_path.to_string(),
+ test_vector_table_schema(vector_file_format),
+ None,
+ );
+
+ let mut table_write = TableWrite::new(&table,
"test-user".to_string()).unwrap();
+ let batch = make_vector_batch(
+ vec![1, 2, 3],
+ vec![vec![1.0, 0.0], vec![0.0, 1.0], vec![0.5, 0.5]],
+ );
+
+ table_write.write_arrow_batch(&batch).await.unwrap();
+ let messages = table_write.prepare_commit().await.unwrap();
+
+ assert_eq!(messages.len(), 1);
+ assert_eq!(messages[0].new_files.len(), 2);
+
+ let vector_suffix = format!(".vector.{vector_file_format}");
+ let normal_files: Vec<_> = messages[0]
+ .new_files
+ .iter()
+ .filter(|f| !f.file_name.contains(".vector."))
+ .collect();
+ let vector_files: Vec<_> = messages[0]
+ .new_files
+ .iter()
+ .filter(|f| f.file_name.ends_with(&vector_suffix))
+ .collect();
+
+ assert_eq!(normal_files.len(), 1);
+ assert_eq!(vector_files.len(), 1);
+ assert_eq!(normal_files[0].row_count, 3);
+ assert_eq!(vector_files[0].row_count, 3);
+ assert_eq!(
+ vector_files[0].write_cols,
+ Some(vec!["embedding".to_string()])
+ );
+
+ TableCommit::new(table, "test-user".to_string())
+ .commit(messages)
+ .await
+ .unwrap();
+ let snap_manager = SnapshotManager::new(file_io.clone(),
table_path.to_string());
+ let snapshot =
snap_manager.get_latest_snapshot().await.unwrap().unwrap();
+ assert_eq!(snapshot.id(), 1);
+ }
+
+ #[tokio::test]
+ async fn test_vector_write_uses_dedicated_parquet_file() {
+ assert_vector_write_uses_dedicated_file(
+ "memory:/test_vector_write_dedicated_parquet",
+ "parquet",
+ )
+ .await;
+ }
+
+ #[cfg(feature = "vortex")]
+ #[tokio::test]
+ async fn test_vector_write_uses_dedicated_vortex_file() {
+ assert_vector_write_uses_dedicated_file(
+ "memory:/test_vector_write_dedicated_vortex",
+ "vortex",
+ )
+ .await;
+ }
+
#[test]
fn test_allows_partial_update_fixed_bucket_table() {
let table = Table::new(
