You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lx...@apache.org on 2026/10/21 10:05:48 UTC
(paimon-cpp) branch main updated: feat(vector): add dedicated storage for append-only tables (#438)
This is an automated email from the ASF dual-hosted git repository.
lxy-9602 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 50ef548e feat(vector): add dedicated storage for append-only tables (#438)
50ef548e is described below
commit 50ef548ec87cc444a153fb2a03789f9be768d610
Author: 小明同学 <72...@users.noreply.github.com>
AuthorDate: Wed Oct 21 18:05:42 2026 +0800
feat(vector): add dedicated storage for append-only tables (#438)
---
docs/source/user_guide/data_types.rst | 22 +-
include/paimon/defs.h | 8 +
src/paimon/CMakeLists.txt | 3 +
src/paimon/common/defs.cpp | 2 +
src/paimon/core/append/append_only_writer.cpp | 32 ++
src/paimon/core/core_options.cpp | 19 +
src/paimon/core/core_options.h | 2 +
src/paimon/core/core_options_test.cpp | 7 +
src/paimon/core/io/data_file_path_factory.h | 4 +
src/paimon/core/io/rolling_vector_file_writer.cpp | 177 ++++++++
src/paimon/core/io/rolling_vector_file_writer.h | 58 +++
.../core/io/rolling_vector_file_writer_test.cpp | 303 ++++++++++++++
.../core/io/vector_data_file_writer_factory.cpp | 49 +++
.../core/io/vector_data_file_writer_factory.h | 34 ++
.../operation/data_evolution_file_store_scan.cpp | 48 ++-
.../data_evolution_file_store_scan_test.cpp | 156 ++++++-
.../core/operation/data_evolution_split_read.cpp | 264 +++++++++++-
.../core/operation/data_evolution_split_read.h | 23 +-
.../operation/data_evolution_split_read_test.cpp | 218 ++++++++++
src/paimon/core/schema/schema_validation.cpp | 61 ++-
src/paimon/core/schema/schema_validation.h | 3 +
src/paimon/core/schema/schema_validation_test.cpp | 103 ++++-
src/paimon/core/table/table.cpp | 5 +
.../format/vortex/vortex_file_format_test.cpp | 133 ++++++
src/paimon/format/vortex/vortex_format_writer.cpp | 86 +++-
src/paimon/format/vortex/vortex_format_writer.h | 1 +
test/inte/blob_table_inte_test.cpp | 21 +
test/inte/data_evolution_table_test.cpp | 458 +++++++++++++++++++++
test/inte/write_and_read_inte_test.cpp | 76 ++++
.../dedicated_vector_java_compat/README.md | 39 ++
...-02c40850-170e-4085-bc42-a69c6d6fffdc-0.parquet | Bin 0 -> 475 bytes
...850-170e-4085-bc42-a69c6d6fffdc-1.vector.vortex | Bin 0 -> 2436 bytes
...-bd5ba451-e0e6-4a75-8a36-2ea3d2387dbe-0.parquet | Bin 0 -> 476 bytes
...451-e0e6-4a75-8a36-2ea3d2387dbe-1.vector.vortex | Bin 0 -> 2436 bytes
...manifest-16ae6a76-0eec-45b3-a1e2-f18249238804-0 | Bin 0 -> 2254 bytes
...manifest-736f9660-1f51-41e2-b0bf-751a67514c1d-0 | Bin 0 -> 2253 bytes
...est-list-d1d44abf-1ad2-4f22-bd0d-bc8b0d1ab444-0 | Bin 0 -> 1261 bytes
...est-list-d1d44abf-1ad2-4f22-bd0d-bc8b0d1ab444-1 | Bin 0 -> 1263 bytes
...est-list-d7544098-f6d3-4754-8d1c-cc9e672f1970-0 | Bin 0 -> 1158 bytes
...est-list-d7544098-f6d3-4754-8d1c-cc9e672f1970-1 | Bin 0 -> 1261 bytes
.../dedicated_vector_java_compat/schema/schema-0 | 33 ++
.../dedicated_vector_java_compat/snapshot/LATEST | 1 +
.../snapshot/snapshot-1 | 18 +
.../snapshot/snapshot-2 | 18 +
.../dedicated_vector_java_rolling/README.md | 46 +++
...-253f5756-e2a8-4ab4-92f0-c0896ef10d06-0.parquet | Bin 0 -> 8814 bytes
...756-e2a8-4ab4-92f0-c0896ef10d06-1.vector.vortex | Bin 0 -> 11540 bytes
...756-e2a8-4ab4-92f0-c0896ef10d06-2.vector.vortex | Bin 0 -> 8420 bytes
...756-e2a8-4ab4-92f0-c0896ef10d06-3.vector.vortex | Bin 0 -> 8420 bytes
...756-e2a8-4ab4-92f0-c0896ef10d06-4.vector.vortex | Bin 0 -> 2148 bytes
...manifest-af194a04-83f3-4b43-a88b-10f8e3d36dd9-0 | Bin 0 -> 2326 bytes
...est-list-947870e6-2276-4657-a9fc-ac454ea83ffd-0 | Bin 0 -> 1158 bytes
...est-list-947870e6-2276-4657-a9fc-ac454ea83ffd-1 | Bin 0 -> 1262 bytes
.../dedicated_vector_java_rolling/schema/schema-0 | 37 ++
.../dedicated_vector_java_rolling/snapshot/LATEST | 1 +
.../snapshot/snapshot-1 | 18 +
56 files changed, 2536 insertions(+), 51 deletions(-)
diff --git a/docs/source/user_guide/data_types.rst b/docs/source/user_guide/data_types.rst
index f864596d..34818ea4 100644
--- a/docs/source/user_guide/data_types.rst
+++ b/docs/source/user_guide/data_types.rst
@@ -197,13 +197,27 @@ and `Arrow DataTypes <https://arrow.apache.org/docs/format/Columnar.html#data-ty
Paimon C++ supports VECTOR value columns in append-only and primary-key
tables backed by Parquet data files. They use the standard Parquet LIST
representation on disk and are restored as Arrow ``FixedSizeList``
- values on read. Row-tracking append-only tables with data evolution also
- support full-row and partial-column VECTOR writes, including reads across
+ values on read. With Vortex enabled at build time, Vortex data files also
+ support VECTOR value columns. Row-tracking append-only Parquet tables
+ with data evolution support full-row and partial-column VECTOR writes, including reads across
schema versions that add or drop VECTOR columns. Changing a VECTOR's
dimension or element type is not supported. VECTOR columns cannot be
primary, partition, or bucket keys, nor comparator-based
- ordering fields such as sequence and sequence-group fields. Dedicated
- vector storage is not included yet.
+ ordering fields such as sequence and sequence-group fields.
+
+ ``vector.file.format`` stores a single top-level VECTOR column in dedicated
+ ``.vector.<format>`` files. Supported formats are ``parquet``, ``lance``,
+ and ``vortex``; Lance and Vortex must be enabled at build time.
+ This mode requires ``file.format=parquet``, ``bucket=-1``,
+ ``row-tracking.enabled=true``, and ``data-evolution.enabled=true``, plus at
+ least one ordinary non-partition column. It supports full-row inserts into append-only
+ tables. Custom write schemas, partial writes, multiple or nested VECTOR
+ columns, and BLOB columns are not supported in this mode.
+
+ VECTOR files roll independently at ``vector.target-file-size``, which
+ defaults to ``target-file-size``. Main files roll at ``target-file-size``
+ or ``target-file-row-num`` and also close the current VECTOR file. One
+ main file can therefore cover the combined row range of several VECTOR files.
Paimon C++ also reads Parquet files written by Paimon Rust or Python whose
embedded Arrow schema restores VECTOR columns as ``FixedSizeList``,
diff --git a/include/paimon/defs.h b/include/paimon/defs.h
index 8c8e3235..709d4490 100644
--- a/include/paimon/defs.h
+++ b/include/paimon/defs.h
@@ -162,6 +162,14 @@ struct PAIMON_EXPORT Options {
/// "blob.target-file-size" - Target size of a blob file. Default is TARGET_FILE_SIZE.
static const char BLOB_TARGET_FILE_SIZE[];
+ /// "vector.file.format" - Optional dedicated VECTOR storage format (parquet, lance or vortex)
+ /// for a single top-level VECTOR column in row-tracking append-only tables.
+ static const char VECTOR_FILE_FORMAT[];
+
+ /// "vector.target-file-size" - Target size of the vector file. Defaults to TARGET_FILE_SIZE.
+ /// VECTOR files roll independently; rolling a main file also closes the current VECTOR file.
+ static const char VECTOR_TARGET_FILE_SIZE[];
+
/// "blob.split-by-file-size" - Whether to consider blob file size as a factor when performing
/// scan splitting. When unset, defaults to the negation of BLOB_AS_DESCRIPTOR.
static const char BLOB_SPLIT_BY_FILE_SIZE[];
diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt
index a8a4afe4..da5213dc 100644
--- a/src/paimon/CMakeLists.txt
+++ b/src/paimon/CMakeLists.txt
@@ -317,6 +317,8 @@ set(PAIMON_CORE_SRCS
core/io/shredding_key_value_data_file_writer_factory.cpp
core/io/multiple_blob_file_writer.cpp
core/io/rolling_blob_file_writer.cpp
+ core/io/rolling_vector_file_writer.cpp
+ core/io/vector_data_file_writer_factory.cpp
core/manifest/file_kind.cpp
core/manifest/file_source.cpp
core/manifest/index_manifest_entry_serializer.cpp
@@ -859,6 +861,7 @@ if(PAIMON_BUILD_TESTS)
core/io/file_index_evaluator_test.cpp
core/io/single_file_writer_test.cpp
core/io/rolling_blob_file_writer_test.cpp
+ core/io/rolling_vector_file_writer_test.cpp
core/global_index/global_index_file_manager_test.cpp
core/global_index/global_index_evaluator_impl_test.cpp
core/global_index/indexed_split_test.cpp
diff --git a/src/paimon/common/defs.cpp b/src/paimon/common/defs.cpp
index 951394e5..d709d6f8 100644
--- a/src/paimon/common/defs.cpp
+++ b/src/paimon/common/defs.cpp
@@ -47,6 +47,8 @@ const char Options::FILE_SYSTEM[] = "file-system";
const char Options::TARGET_FILE_SIZE[] = "target-file-size";
const char Options::TARGET_FILE_ROW_NUM[] = "target-file-row-num";
const char Options::BLOB_TARGET_FILE_SIZE[] = "blob.target-file-size";
+const char Options::VECTOR_FILE_FORMAT[] = "vector.file.format";
+const char Options::VECTOR_TARGET_FILE_SIZE[] = "vector.target-file-size";
const char Options::BLOB_SPLIT_BY_FILE_SIZE[] = "blob.split-by-file-size";
const char Options::PAGE_SIZE[] = "page-size";
const char Options::PARTITION_DEFAULT_NAME[] = "partition.default-name";
diff --git a/src/paimon/core/append/append_only_writer.cpp b/src/paimon/core/append/append_only_writer.cpp
index 45d4e29d..9b3f32c5 100644
--- a/src/paimon/core/append/append_only_writer.cpp
+++ b/src/paimon/core/append/append_only_writer.cpp
@@ -33,6 +33,7 @@
#include "paimon/common/metrics/metrics_impl.h"
#include "paimon/common/types/row_kind.h"
#include "paimon/common/utils/arrow/status_utils.h"
+#include "paimon/common/utils/arrow/vector_utils.h"
#include "paimon/common/utils/checked_cast.h"
#include "paimon/common/utils/long_counter.h"
#include "paimon/common/utils/scope_guard.h"
@@ -44,8 +45,10 @@
#include "paimon/core/io/multiple_blob_file_writer.h"
#include "paimon/core/io/rolling_blob_file_writer.h"
#include "paimon/core/io/rolling_file_writer.h"
+#include "paimon/core/io/rolling_vector_file_writer.h"
#include "paimon/core/io/shredding_append_data_file_writer_factory.h"
#include "paimon/core/io/single_file_writer.h"
+#include "paimon/core/io/vector_data_file_writer_factory.h"
#include "paimon/core/manifest/file_source.h"
#include "paimon/core/operation/blob_file_context.h"
#include "paimon/core/utils/commit_increment.h"
@@ -176,6 +179,35 @@ Status AppendOnlyWriter::Flush(bool wait_for_latest_compaction, bool forced_full
}
AppendOnlyWriter::RollingFileWriterResult AppendOnlyWriter::CreateRollingRowWriter() {
+ if (options_.GetVectorFileFormat()) {
+ arrow::FieldVector main_fields;
+ arrow::FieldVector vector_fields;
+ int32_t vector_index = -1;
+ for (int32_t i = 0; i < write_schema_->num_fields(); ++i) {
+ const auto& field = write_schema_->field(i);
+ if (VectorUtils::ContainsVectorField(field)) {
+ vector_fields.push_back(field);
+ vector_index = i;
+ } else {
+ main_fields.push_back(field);
+ }
+ }
+ if (vector_fields.size() != 1 || main_fields.empty()) {
+ return Status::Invalid(
+ "Dedicated VECTOR storage requires normal columns and one VECTOR column");
+ }
+ auto main_schema = arrow::schema(main_fields);
+ auto vector_schema = arrow::schema(vector_fields);
+ PAIMON_ASSIGN_OR_RAISE(WriterFactory main_factory,
+ GetDataFileWriterFactory(main_schema, main_schema->field_names()));
+ auto vector_factory = std::make_shared<VectorDataFileWriterFactory>(
+ options_, schema_id_, vector_schema, vector_schema->field_names(), seq_num_counter_,
+ FileSource::Append(), path_factory_, memory_pool_);
+ return std::make_unique<RollingVectorFileWriter>(
+ options_.GetTargetFileSize(/*has_primary_key=*/false), options_.GetTargetFileRowNum(),
+ options_.GetVectorTargetFileSize(), main_factory, vector_factory, write_schema_,
+ vector_index);
+ }
auto blob_context = BlobFileContext::Create(write_schema_, options_);
// Save inline descriptor and view fields for validation in Write()
diff --git a/src/paimon/core/core_options.cpp b/src/paimon/core/core_options.cpp
index 92afe6ea..666d7685 100644
--- a/src/paimon/core/core_options.cpp
+++ b/src/paimon/core/core_options.cpp
@@ -358,7 +358,9 @@ struct CoreOptions::Impl {
std::optional<int64_t> target_file_size;
std::optional<int64_t> blob_target_file_size;
+ std::optional<int64_t> vector_target_file_size;
std::shared_ptr<FileFormat> file_format;
+ std::shared_ptr<FileFormat> vector_file_format;
std::shared_ptr<FileSystem> file_system;
std::shared_ptr<FileFormat> manifest_file_format;
std::shared_ptr<Cache> cache;
@@ -599,6 +601,15 @@ struct CoreOptions::Impl {
// Parse file.format - data file format, default "parquet"
PAIMON_RETURN_NOT_OK(parser.ParseObject<FileFormatFactory>(
Options::FILE_FORMAT, /*default_identifier=*/"parquet", &file_format));
+ if (parser.ContainsKey(Options::VECTOR_FILE_FORMAT)) {
+ PAIMON_RETURN_NOT_OK(parser.ParseObject<FileFormatFactory>(
+ Options::VECTOR_FILE_FORMAT, /*default_identifier=*/"vortex", &vector_file_format));
+ }
+ PAIMON_RETURN_NOT_OK(
+ parser.ParseMemorySize(Options::VECTOR_TARGET_FILE_SIZE, &vector_target_file_size));
+ if (vector_target_file_size && vector_target_file_size.value() <= 0) {
+ return Status::Invalid("vector.target-file-size must be positive");
+ }
// Parse file.compression - default file compression, default "zstd"
PAIMON_RETURN_NOT_OK(parser.Parse(Options::FILE_COMPRESSION, &file_compression));
// Parse changelog-file.format - no default value
@@ -1123,6 +1134,10 @@ std::shared_ptr<FileFormat> CoreOptions::GetFileFormat() const {
return impl_->file_format;
}
+std::shared_ptr<FileFormat> CoreOptions::GetVectorFileFormat() const {
+ return impl_->vector_file_format;
+}
+
std::shared_ptr<FileSystem> CoreOptions::GetFileSystem() const {
return impl_->file_system;
}
@@ -1169,6 +1184,10 @@ bool CoreOptions::BlobSplitByFileSize() const {
return impl_->blob_split_by_file_size.value_or(!impl_->blob_as_descriptor);
}
+int64_t CoreOptions::GetVectorTargetFileSize() const {
+ return impl_->vector_target_file_size.value_or(GetTargetFileSize(/*has_primary_key=*/false));
+}
+
int64_t CoreOptions::GetCompactionFileSize(bool has_primary_key) const {
// file size to join the compaction, we don't process on middle file size to avoid
// compact a same file twice (the compression is not calculate so accurately. the output
diff --git a/src/paimon/core/core_options.h b/src/paimon/core/core_options.h
index a16f5e67..8bf79426 100644
--- a/src/paimon/core/core_options.h
+++ b/src/paimon/core/core_options.h
@@ -80,6 +80,7 @@ class PAIMON_EXPORT CoreOptions {
int32_t GetBucket() const;
std::shared_ptr<FileFormat> GetFileFormat() const;
+ std::shared_ptr<FileFormat> GetVectorFileFormat() const;
std::shared_ptr<FileFormat> GetWriteFileFormat(int32_t level) const;
std::shared_ptr<FileSystem> GetFileSystem() const;
const std::string& GetFileCompression() const;
@@ -89,6 +90,7 @@ class PAIMON_EXPORT CoreOptions {
int64_t GetTargetFileSize(bool has_primary_key) const;
int64_t GetTargetFileRowNum() const;
int64_t GetBlobTargetFileSize() const;
+ int64_t GetVectorTargetFileSize() const;
bool BlobSplitByFileSize() const;
int64_t GetCompactionFileSize(bool has_primary_key) const;
std::string GetPartitionDefaultName() const;
diff --git a/src/paimon/core/core_options_test.cpp b/src/paimon/core/core_options_test.cpp
index 074105cd..e3625db8 100644
--- a/src/paimon/core/core_options_test.cpp
+++ b/src/paimon/core/core_options_test.cpp
@@ -53,6 +53,7 @@ TEST(CoreOptionsTest, TestDefaultValue) {
ASSERT_EQ(manifest_format->Identifier(), "avro");
ASSERT_EQ(core_options.GetFileFormat()->Identifier(), "parquet");
ASSERT_EQ(nullptr, core_options.GetChangelogFileFormat());
+ ASSERT_EQ(nullptr, core_options.GetVectorFileFormat());
ASSERT_EQ(core_options.GetWriteFileFormat(0)->Identifier(), "parquet");
ASSERT_EQ(core_options.GetWriteFileFormat(3)->Identifier(), "parquet");
ASSERT_TRUE(core_options.GetFileSystem());
@@ -62,6 +63,7 @@ TEST(CoreOptionsTest, TestDefaultValue) {
ASSERT_EQ(128 * 1024 * 1024L, core_options.GetTargetFileSize(/*has_primary_key=*/true));
ASSERT_EQ(std::numeric_limits<int64_t>::max(), core_options.GetTargetFileRowNum());
ASSERT_EQ(256 * 1024 * 1024L, core_options.GetBlobTargetFileSize());
+ ASSERT_EQ(256 * 1024 * 1024L, core_options.GetVectorTargetFileSize());
ASSERT_TRUE(core_options.BlobSplitByFileSize());
ASSERT_EQ(187904815, core_options.GetCompactionFileSize(/*has_primary_key=*/false));
ASSERT_EQ(93952404, core_options.GetCompactionFileSize(/*has_primary_key=*/true));
@@ -214,12 +216,14 @@ TEST(CoreOptionsTest, TestFromMap) {
{Options::FILE_SYSTEM, "Local"},
{Options::FILE_FORMAT, "ORC"},
{Options::CHANGELOG_FILE_FORMAT, "avro"},
+ {Options::VECTOR_FILE_FORMAT, "parquet"},
{"manifest.format", "ORC"},
{Options::BUCKET, "3"},
{Options::PAGE_SIZE, "128 kb"},
{Options::TARGET_FILE_SIZE, "512MB"},
{Options::TARGET_FILE_ROW_NUM, "123"},
{Options::BLOB_TARGET_FILE_SIZE, "1G"},
+ {Options::VECTOR_TARGET_FILE_SIZE, "2MB"},
{Options::PARTITION_DEFAULT_NAME, "foo"},
{Options::MANIFEST_TARGET_FILE_SIZE, "16MB"},
{Options::MANIFEST_FULL_COMPACTION_FILE_SIZE, "32MB"},
@@ -354,6 +358,8 @@ TEST(CoreOptionsTest, TestFromMap) {
ASSERT_EQ(core_options.GetFileFormat()->Identifier(), "orc");
ASSERT_EQ(core_options.GetChangelogFileFormat()->Identifier(), "avro");
+ ASSERT_NE(nullptr, core_options.GetVectorFileFormat());
+ ASSERT_EQ(core_options.GetVectorFileFormat()->Identifier(), "parquet");
ASSERT_EQ(core_options.GetWriteFileFormat(0)->Identifier(), "avro");
ASSERT_EQ(core_options.GetWriteFileFormat(1)->Identifier(), "orc");
ASSERT_EQ(core_options.GetWriteFileFormat(3)->Identifier(), "parquet");
@@ -367,6 +373,7 @@ TEST(CoreOptionsTest, TestFromMap) {
ASSERT_EQ(512 * 1024 * 1024L, core_options.GetTargetFileSize(/*has_primary_key=*/false));
ASSERT_EQ(123, core_options.GetTargetFileRowNum());
ASSERT_EQ(1024 * 1024 * 1024L, core_options.GetBlobTargetFileSize());
+ ASSERT_EQ(2 * 1024 * 1024L, core_options.GetVectorTargetFileSize());
ASSERT_EQ("foo", core_options.GetPartitionDefaultName());
ASSERT_EQ(16 * 1024 * 1024L, core_options.GetManifestTargetFileSize());
ASSERT_EQ(32 * 1024 * 1024L, core_options.GetManifestFullCompactionThresholdSize());
diff --git a/src/paimon/core/io/data_file_path_factory.h b/src/paimon/core/io/data_file_path_factory.h
index a9050eee..54f4168e 100644
--- a/src/paimon/core/io/data_file_path_factory.h
+++ b/src/paimon/core/io/data_file_path_factory.h
@@ -65,6 +65,10 @@ class DataFilePathFactory : public PathFactory {
return NewPathFromName(NewFileName(data_file_prefix_, ".blob"));
}
+ std::string NewVectorPath(const std::string& format_identifier) const {
+ return NewPathFromName(NewFileName(data_file_prefix_, ".vector." + format_identifier));
+ }
+
std::string NewPathFromName(const std::string& file_name) const {
if (external_path_provider_ != nullptr) {
return external_path_provider_->GetNextExternalDataPath(file_name);
diff --git a/src/paimon/core/io/rolling_vector_file_writer.cpp b/src/paimon/core/io/rolling_vector_file_writer.cpp
new file mode 100644
index 00000000..387facec
--- /dev/null
+++ b/src/paimon/core/io/rolling_vector_file_writer.cpp
@@ -0,0 +1,177 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "paimon/core/io/rolling_vector_file_writer.h"
+
+#include "arrow/array/array_nested.h"
+#include "paimon/common/utils/arrow/status_utils.h"
+#include "paimon/common/utils/checked_cast.h"
+
+namespace paimon {
+
+RollingVectorFileWriter::RollingVectorFileWriter(
+ int64_t target_file_size, int64_t target_file_row_num, int64_t vector_target_file_size,
+ const std::shared_ptr<WriterFactory>& main_factory,
+ const std::shared_ptr<WriterFactory>& vector_factory,
+ const std::shared_ptr<arrow::Schema>& write_schema, int32_t vector_index)
+ : RollingFileWriter(target_file_size, target_file_row_num, main_factory),
+ vector_target_file_size_(vector_target_file_size),
+ vector_factory_(vector_factory),
+ write_schema_(write_schema),
+ vector_index_(vector_index) {}
+
+Status RollingVectorFileWriter::Write(::ArrowArray* record) {
+ if (closed_ || aborted_) {
+ return Status::Invalid("Writer has already closed or aborted!");
+ }
+ ScopeGuard guard([this]() { Abort(); });
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ std::shared_ptr<arrow::Array> array,
+ arrow::ImportArray(record, arrow::struct_(write_schema_->fields())));
+ if (!array || array->type_id() != arrow::Type::STRUCT || array->null_count() != 0) {
+ return Status::Invalid("Dedicated VECTOR storage expects non-null struct rows");
+ }
+ if (array->length() == 0) {
+ guard.Release();
+ return Status::OK();
+ }
+ if (!current_writer_) {
+ PAIMON_RETURN_NOT_OK(OpenCurrentWriter());
+ }
+ if (!vector_writer_) {
+ PAIMON_ASSIGN_OR_RAISE(vector_writer_, vector_factory_->CreateWriter());
+ }
+ auto rows = checked_pointer_cast<arrow::StructArray>(array);
+ arrow::ArrayVector main_arrays;
+ arrow::FieldVector main_fields;
+ for (int32_t i = 0; i < write_schema_->num_fields(); ++i) {
+ if (i != vector_index_) {
+ main_arrays.push_back(rows->field(i));
+ main_fields.push_back(write_schema_->field(i));
+ }
+ }
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::StructArray> main_array,
+ arrow::StructArray::Make(main_arrays, main_fields));
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ std::shared_ptr<arrow::StructArray> vector_array,
+ arrow::StructArray::Make({rows->field(vector_index_)},
+ {write_schema_->field(vector_index_)}));
+ ::ArrowArray main_batch = {};
+ ::ArrowArray vector_batch = {};
+ ScopeGuard arrays_guard([&]() {
+ ArrowArrayRelease(&main_batch);
+ ArrowArrayRelease(&vector_batch);
+ });
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*main_array, &main_batch));
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*vector_array, &vector_batch));
+ PAIMON_RETURN_NOT_OK(current_writer_->Write(&main_batch));
+ PAIMON_RETURN_NOT_OK(vector_writer_->Write(&vector_batch));
+ record_count_ += rows->length();
+ current_file_record_count_ += rows->length();
+
+ const bool suggested_check = SuggestCheck();
+ PAIMON_ASSIGN_OR_RAISE(bool main_full,
+ current_writer_->ReachTargetSize(suggested_check, target_file_size_));
+ PAIMON_ASSIGN_OR_RAISE(bool vector_full, vector_writer_->ReachTargetSize(
+ suggested_check, vector_target_file_size_));
+ if (main_full || current_file_record_count_ >= target_file_row_num_) {
+ PAIMON_RETURN_NOT_OK(CloseCurrentMainWriter());
+ } else if (vector_full) {
+ PAIMON_RETURN_NOT_OK(CloseCurrentVectorWriter());
+ }
+ guard.Release();
+ return Status::OK();
+}
+
+Status RollingVectorFileWriter::CloseCurrentVectorWriter() {
+ if (!vector_writer_) {
+ return Status::OK();
+ }
+ PAIMON_RETURN_NOT_OK(vector_writer_->Close());
+ PAIMON_ASSIGN_OR_RAISE(Writer::AbortExecutor vector_abort, vector_writer_->GetAbortExecutor());
+ closed_writers_.push_back(vector_abort);
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<DataFileMeta> vector_meta, vector_writer_->GetResult());
+ current_vector_files_.push_back(vector_meta);
+ metrics_->Merge(vector_writer_->GetMetrics());
+ vector_writer_.reset();
+ return Status::OK();
+}
+
+Status RollingVectorFileWriter::CloseCurrentMainWriter() {
+ if (!current_writer_) {
+ return Status::OK();
+ }
+ PAIMON_RETURN_NOT_OK(current_writer_->Close());
+ PAIMON_ASSIGN_OR_RAISE(Writer::AbortExecutor main_abort, current_writer_->GetAbortExecutor());
+ closed_writers_.push_back(main_abort);
+ PAIMON_RETURN_NOT_OK(CloseCurrentVectorWriter());
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<DataFileMeta> main_meta, current_writer_->GetResult());
+ int64_t vector_row_count = 0;
+ for (const auto& vector_meta : current_vector_files_) {
+ vector_row_count += vector_meta->row_count;
+ }
+ if (main_meta->row_count != vector_row_count) {
+ return Status::Invalid("Main and VECTOR files must contain the same total number of rows");
+ }
+ // Row tracking assigns ids in manifest order: emit the main file before its VECTOR files.
+ results_.push_back(main_meta);
+ results_.insert(results_.end(), current_vector_files_.begin(), current_vector_files_.end());
+ current_vector_files_.clear();
+ metrics_->Merge(current_writer_->GetMetrics());
+ current_writer_.reset();
+ current_file_record_count_ = 0;
+ return Status::OK();
+}
+
+Status RollingVectorFileWriter::Close() {
+ if (aborted_) {
+ return Status::Invalid("Writer has been aborted!");
+ }
+ if (closed_) {
+ return Status::OK();
+ }
+ Status status = CloseCurrentMainWriter();
+ if (!status.ok()) {
+ Abort();
+ }
+ closed_ = true;
+ return status;
+}
+
+void RollingVectorFileWriter::Abort() {
+ aborted_ = true;
+ if (vector_writer_) {
+ vector_writer_->Abort();
+ vector_writer_.reset();
+ }
+ RollingFileWriter::Abort();
+ current_writer_.reset();
+ current_vector_files_.clear();
+ closed_writers_.clear();
+ results_.clear();
+}
+
+Result<std::vector<std::shared_ptr<DataFileMeta>>> RollingVectorFileWriter::GetResult() {
+ if (aborted_) {
+ return Status::Invalid("Writer has been aborted!");
+ }
+ return RollingFileWriter::GetResult();
+}
+
+} // namespace paimon
diff --git a/src/paimon/core/io/rolling_vector_file_writer.h b/src/paimon/core/io/rolling_vector_file_writer.h
new file mode 100644
index 00000000..c20dc69e
--- /dev/null
+++ b/src/paimon/core/io/rolling_vector_file_writer.h
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#pragma once
+
+#include "paimon/core/io/rolling_file_writer.h"
+
+namespace paimon {
+
+/// Writes normal columns and one VECTOR column, rolling VECTOR files independently.
+class RollingVectorFileWriter
+ : public RollingFileWriter<::ArrowArray*, std::shared_ptr<DataFileMeta>> {
+ public:
+ using Writer = SingleFileWriter<::ArrowArray*, std::shared_ptr<DataFileMeta>>;
+ using WriterFactory = SingleFileWriterFactory<::ArrowArray*, std::shared_ptr<DataFileMeta>>;
+
+ RollingVectorFileWriter(int64_t target_file_size, int64_t target_file_row_num,
+ int64_t vector_target_file_size,
+ const std::shared_ptr<WriterFactory>& main_factory,
+ const std::shared_ptr<WriterFactory>& vector_factory,
+ const std::shared_ptr<arrow::Schema>& write_schema,
+ int32_t vector_index);
+
+ Status Write(::ArrowArray* record) override;
+ Status Close() override;
+ void Abort() override;
+ Result<std::vector<std::shared_ptr<DataFileMeta>>> GetResult() override;
+
+ private:
+ Status CloseCurrentVectorWriter();
+ Status CloseCurrentMainWriter();
+
+ int64_t vector_target_file_size_;
+ std::shared_ptr<WriterFactory> vector_factory_;
+ std::shared_ptr<arrow::Schema> write_schema_;
+ int32_t vector_index_;
+ std::unique_ptr<Writer> vector_writer_;
+ std::vector<std::shared_ptr<DataFileMeta>> current_vector_files_;
+ bool aborted_ = false;
+};
+
+} // namespace paimon
diff --git a/src/paimon/core/io/rolling_vector_file_writer_test.cpp b/src/paimon/core/io/rolling_vector_file_writer_test.cpp
new file mode 100644
index 00000000..c65e2d71
--- /dev/null
+++ b/src/paimon/core/io/rolling_vector_file_writer_test.cpp
@@ -0,0 +1,303 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "paimon/core/io/rolling_vector_file_writer.h"
+
+#include <limits>
+#include <tuple>
+#include <utility>
+
+#include "arrow/ipc/json_simple.h"
+#include "gtest/gtest.h"
+#include "paimon/common/utils/path_util.h"
+#include "paimon/core/manifest/file_source.h"
+#include "paimon/core/stats/simple_stats.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+namespace {
+
+struct WriterState {
+ bool fail_create = false;
+ bool fail_write = false;
+ bool fail_close = false;
+ bool fail_size = false;
+ bool fail_result = false;
+ int64_t row_count_delta = 0;
+ int64_t file_number = 0;
+ std::vector<std::string> paths;
+};
+
+class TestVectorWriter : public RollingVectorFileWriter::Writer {
+ public:
+ TestVectorWriter(const std::shared_ptr<FileSystem>& fs, const std::string& path,
+ const std::shared_ptr<WriterState>& state)
+ : SingleFileWriter("none", {}), state_(state) {
+ fs_ = fs;
+ path_ = path;
+ }
+
+ Status Write(::ArrowArray* record) override {
+ ScopeGuard guard([&]() { ArrowArrayRelease(record); });
+ if (state_->fail_write) {
+ return Status::IOError("injected write failure");
+ }
+ rows_ += record->length;
+ return Status::OK();
+ }
+
+ Status Close() override {
+ if (state_->fail_close) {
+ return Status::IOError("injected close failure");
+ }
+ closed_ = true;
+ return Status::OK();
+ }
+
+ Result<bool> ReachTargetSize(bool, int64_t target_size) override {
+ if (state_->fail_size) {
+ return Status::IOError("injected size check failure");
+ }
+ return rows_ >= target_size;
+ }
+
+ Result<std::shared_ptr<DataFileMeta>> GetResult() override {
+ if (state_->fail_result) {
+ return Status::IOError("injected result failure");
+ }
+ return DataFileMeta::ForAppend(PathUtil::GetName(path_), 0, rows_ + state_->row_count_delta,
+ SimpleStats::EmptyStats(), 0, rows_ - 1, 0,
+ FileSource::Append(), std::nullopt, std::nullopt,
+ std::nullopt, std::nullopt);
+ }
+
+ private:
+ std::shared_ptr<WriterState> state_;
+ int64_t rows_ = 0;
+};
+
+class TestVectorWriterFactory : public RollingVectorFileWriter::WriterFactory {
+ public:
+ TestVectorWriterFactory(std::shared_ptr<FileSystem> fs, std::string prefix,
+ std::shared_ptr<WriterState> state)
+ : fs_(std::move(fs)), prefix_(std::move(prefix)), state_(std::move(state)) {}
+
+ Result<std::unique_ptr<RollingVectorFileWriter::Writer>> CreateWriter() const override {
+ if (state_->fail_create) {
+ return Status::IOError("injected create failure");
+ }
+ std::string path = fmt::format("{}{}", prefix_, state_->file_number++);
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<OutputStream> stream,
+ fs_->Create(path, /*overwrite=*/false));
+ PAIMON_RETURN_NOT_OK(stream->Close());
+ state_->paths.push_back(path);
+ return std::make_unique<TestVectorWriter>(fs_, path, state_);
+ }
+
+ private:
+ std::shared_ptr<FileSystem> fs_;
+ std::string prefix_;
+ std::shared_ptr<WriterState> state_;
+};
+
+} // namespace
+
+class RollingVectorFileWriterTest : public ::testing::Test {
+ protected:
+ void SetUp() override {
+ dir_ = UniqueTestDirectory::Create("local");
+ ASSERT_NE(dir_, nullptr);
+ fs_ = dir_->GetFileSystem();
+ main_state_ = std::make_shared<WriterState>();
+ vector_state_ = std::make_shared<WriterState>();
+ schema_ = arrow::schema({arrow::field("id", arrow::int32()),
+ arrow::field("v", arrow::fixed_size_list(arrow::float32(), 2))});
+ }
+
+ std::unique_ptr<RollingVectorFileWriter> NewWriter(
+ int64_t main_size = 1024, int64_t main_rows = std::numeric_limits<int64_t>::max(),
+ int64_t vector_size = 1) const {
+ auto main_factory = std::make_shared<TestVectorWriterFactory>(
+ fs_, PathUtil::JoinPath(dir_->Str(), "main"), main_state_);
+ auto vector_factory = std::make_shared<TestVectorWriterFactory>(
+ fs_, PathUtil::JoinPath(dir_->Str(), "vector"), vector_state_);
+ return std::make_unique<RollingVectorFileWriter>(main_size, main_rows, vector_size,
+ main_factory, vector_factory, schema_, 1);
+ }
+
+ Status WriteRows(RollingVectorFileWriter* writer, const std::string& json) const {
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ std::shared_ptr<arrow::Array> array,
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(schema_->fields()), json));
+ ::ArrowArray batch = {};
+ ScopeGuard guard([&]() { ArrowArrayRelease(&batch); });
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*array, &batch));
+ return writer->Write(&batch);
+ }
+
+ void CheckFilesRemoved() const {
+ for (const auto& state : {main_state_, vector_state_}) {
+ for (const auto& path : state->paths) {
+ ASSERT_OK_AND_ASSIGN(bool exists, fs_->Exists(path));
+ ASSERT_FALSE(exists) << path;
+ }
+ }
+ }
+
+ std::unique_ptr<UniqueTestDirectory> dir_;
+ std::shared_ptr<FileSystem> fs_;
+ std::shared_ptr<WriterState> main_state_;
+ std::shared_ptr<WriterState> vector_state_;
+ std::shared_ptr<arrow::Schema> schema_;
+};
+
+TEST_F(RollingVectorFileWriterTest, VectorTargetRollsIndependently) {
+ auto writer = NewWriter(1024, std::numeric_limits<int64_t>::max(), 2);
+ ASSERT_OK(WriteRows(writer.get(), R"([[1, [1, 2]], [2, null]])"));
+ ASSERT_OK(WriteRows(writer.get(), R"([[3, null], [4, [3, 4]]])"));
+ ASSERT_EQ(main_state_->paths.size(), 1);
+ ASSERT_EQ(vector_state_->paths.size(), 2);
+ ASSERT_OK(WriteRows(writer.get(), R"([[5, null]])"));
+ ASSERT_NOK(writer->GetResult());
+ ASSERT_OK(writer->Close());
+ ASSERT_OK(writer->Close());
+ ASSERT_EQ(writer->RecordCount(), 5);
+ ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<DataFileMeta>> results, writer->GetResult());
+ ASSERT_EQ(results.size(), 4);
+ ASSERT_EQ(results[0]->file_name, "main0");
+ ASSERT_EQ(results[0]->row_count, 5);
+ for (int32_t i = 0; i < 3; ++i) {
+ ASSERT_EQ(results[i + 1]->file_name, fmt::format("vector{}", i));
+ ASSERT_EQ(results[i + 1]->row_count, i == 2 ? 1 : 2);
+ }
+ ASSERT_EQ(main_state_->paths.size(), 1);
+ ASSERT_EQ(vector_state_->paths.size(), 3);
+ writer->Abort();
+ CheckFilesRemoved();
+ ASSERT_NOK(writer->GetResult());
+}
+
+TEST_F(RollingVectorFileWriterTest, MainSizeTargetClosesCurrentVector) {
+ auto writer = NewWriter(2, std::numeric_limits<int64_t>::max(), 1024);
+ ASSERT_OK(WriteRows(writer.get(), R"([[1, [1, 2]], [2, null]])"));
+ ASSERT_OK(WriteRows(writer.get(), R"([[3, null]])"));
+ ASSERT_OK(writer->Close());
+ ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<DataFileMeta>> results, writer->GetResult());
+ ASSERT_EQ(results.size(), 4);
+ ASSERT_EQ(results[0]->file_name, "main0");
+ ASSERT_EQ(results[1]->file_name, "vector0");
+ ASSERT_EQ(results[0]->row_count, 2);
+ ASSERT_EQ(results[1]->row_count, 2);
+ ASSERT_EQ(results[2]->file_name, "main1");
+ ASSERT_EQ(results[3]->file_name, "vector1");
+ ASSERT_EQ(results[2]->row_count, 1);
+ ASSERT_EQ(results[3]->row_count, 1);
+}
+
+TEST_F(RollingVectorFileWriterTest, MainRowLimitPreservesVectorOrderAcrossGroups) {
+ auto writer = NewWriter(1024, 3, 2);
+ for (int32_t i = 0; i < 7; ++i) {
+ ASSERT_OK(WriteRows(writer.get(), R"([[1, null]])"));
+ }
+ ASSERT_OK(writer->Close());
+ ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<DataFileMeta>> results, writer->GetResult());
+ const std::vector<std::string> names = {"main0", "vector0", "vector1", "main1",
+ "vector2", "vector3", "main2", "vector4"};
+ const std::vector<int64_t> counts = {3, 2, 1, 3, 2, 1, 1, 1};
+ ASSERT_EQ(results.size(), names.size());
+ for (size_t i = 0; i < results.size(); ++i) {
+ ASSERT_EQ(results[i]->file_name, names[i]);
+ ASSERT_EQ(results[i]->row_count, counts[i]);
+ }
+}
+
+TEST_F(RollingVectorFileWriterTest, EmptyBatchesDoNotCreateFiles) {
+ auto writer = NewWriter();
+ ASSERT_OK(WriteRows(writer.get(), "[]"));
+ ASSERT_OK(writer->Close());
+ ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<DataFileMeta>> results, writer->GetResult());
+ ASSERT_TRUE(results.empty());
+ ASSERT_TRUE(main_state_->paths.empty());
+ ASSERT_TRUE(vector_state_->paths.empty());
+}
+
+TEST_F(RollingVectorFileWriterTest, ExactVectorBoundaryDoesNotCreateEmptyTail) {
+ auto writer = NewWriter();
+ ASSERT_OK(WriteRows(writer.get(), R"([[1, null]])"));
+ ASSERT_OK(WriteRows(writer.get(), "[]"));
+ ASSERT_OK(writer->Close());
+ ASSERT_EQ(main_state_->paths.size(), 1);
+ ASSERT_EQ(vector_state_->paths.size(), 1);
+ ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<DataFileMeta>> results, writer->GetResult());
+ ASSERT_EQ(results.size(), 2);
+ ASSERT_EQ(results[0]->row_count, 1);
+ ASSERT_EQ(results[1]->row_count, 1);
+}
+
+TEST_F(RollingVectorFileWriterTest, AbortRemovesPendingVectorFiles) {
+ auto writer = NewWriter();
+ ASSERT_OK(WriteRows(writer.get(), R"([[1, null]])"));
+ ASSERT_OK(WriteRows(writer.get(), R"([[2, [1, 2]]])"));
+ writer->Abort();
+ CheckFilesRemoved();
+ ASSERT_NOK(writer->Close());
+ ASSERT_NOK(writer->GetResult());
+}
+
+TEST_F(RollingVectorFileWriterTest, RejectsMismatchedTotalRows) {
+ auto writer = NewWriter();
+ vector_state_->row_count_delta = 1;
+ ASSERT_OK(WriteRows(writer.get(), R"([[1, null]])"));
+ ASSERT_NOK_WITH_MSG(writer->Close(), "same total number of rows");
+ CheckFilesRemoved();
+ ASSERT_NOK(writer->GetResult());
+}
+
+class RollingVectorFailureTest : public RollingVectorFileWriterTest,
+ public ::testing::WithParamInterface<std::tuple<bool, int32_t>> {};
+
+TEST_P(RollingVectorFailureTest, RemovesEarlierGroupsAndPendingVectors) {
+ const auto [main_failure, failure] = GetParam();
+ auto writer = NewWriter(2);
+ ASSERT_OK(WriteRows(writer.get(), R"([[1, null]])"));
+ ASSERT_OK(WriteRows(writer.get(), R"([[2, [1, 2]]])"));
+ // Keep a main file open with an already closed VECTOR file, except when testing main create.
+ if (!main_failure || failure != 0) {
+ ASSERT_OK(WriteRows(writer.get(), R"([[3, null]])"));
+ }
+ const auto& state = main_failure ? main_state_ : vector_state_;
+ state->fail_create = failure == 0;
+ state->fail_write = failure == 1;
+ state->fail_close = failure == 2;
+ state->fail_size = failure == 3;
+ state->fail_result = failure == 4;
+ if (main_failure && (failure == 2 || failure == 4)) {
+ ASSERT_NOK(writer->Close());
+ } else {
+ ASSERT_NOK(WriteRows(writer.get(), R"([[4, [3, 4]]])"));
+ }
+ ASSERT_NOK(writer->GetResult());
+ ASSERT_NOK(writer->Close());
+ CheckFilesRemoved();
+}
+
+INSTANTIATE_TEST_SUITE_P(MainOrVector, RollingVectorFailureTest,
+ ::testing::Combine(::testing::Bool(), ::testing::Range(0, 5)));
+
+} // namespace paimon::test
diff --git a/src/paimon/core/io/vector_data_file_writer_factory.cpp b/src/paimon/core/io/vector_data_file_writer_factory.cpp
new file mode 100644
index 00000000..e2812bd7
--- /dev/null
+++ b/src/paimon/core/io/vector_data_file_writer_factory.cpp
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "paimon/core/io/vector_data_file_writer_factory.h"
+
+#include <utility>
+
+#include "paimon/core/io/data_file_path_factory.h"
+#include "paimon/format/file_format.h"
+
+namespace paimon {
+
+Result<std::unique_ptr<SingleFileWriter<::ArrowArray*, std::shared_ptr<DataFileMeta>>>>
+VectorDataFileWriterFactory::CreateWriter() const {
+ const std::shared_ptr<FileFormat> format = options_.GetVectorFileFormat();
+ if (!format) {
+ return Status::Invalid("vector.file.format is required for dedicated VECTOR files");
+ }
+ PAIMON_ASSIGN_OR_RAISE(WriterResources resources,
+ CreateWriterResources(*format, write_schema_,
+ /*create_stats_extractor=*/true));
+ auto writer = std::make_unique<DataFileWriter>(
+ options_.GetFileCompression(), std::function<Status(::ArrowArray*, ::ArrowArray*)>(),
+ schema_id_, ResolveSeqNumCounter(), file_source_, resources.stats_extractor,
+ path_factory_->IsExternalPath(), write_cols_, pool_);
+ PAIMON_RETURN_NOT_OK(writer->Init(options_.GetFileSystem(),
+ path_factory_->NewVectorPath(format->Identifier()),
+ resources.writer_builder));
+ return std::unique_ptr<SingleFileWriter<::ArrowArray*, std::shared_ptr<DataFileMeta>>>(
+ std::move(writer));
+}
+
+} // namespace paimon
diff --git a/src/paimon/core/io/vector_data_file_writer_factory.h b/src/paimon/core/io/vector_data_file_writer_factory.h
new file mode 100644
index 00000000..2f5be176
--- /dev/null
+++ b/src/paimon/core/io/vector_data_file_writer_factory.h
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#pragma once
+
+#include "paimon/core/io/append_data_file_writer_factory.h"
+
+namespace paimon {
+
+class VectorDataFileWriterFactory : public AppendDataFileWriterFactory {
+ public:
+ using AppendDataFileWriterFactory::AppendDataFileWriterFactory;
+
+ Result<std::unique_ptr<SingleFileWriter<::ArrowArray*, std::shared_ptr<DataFileMeta>>>>
+ CreateWriter() const override;
+};
+
+} // namespace paimon
diff --git a/src/paimon/core/operation/data_evolution_file_store_scan.cpp b/src/paimon/core/operation/data_evolution_file_store_scan.cpp
index 3899146a..3ef72899 100644
--- a/src/paimon/core/operation/data_evolution_file_store_scan.cpp
+++ b/src/paimon/core/operation/data_evolution_file_store_scan.cpp
@@ -18,14 +18,16 @@
#include "paimon/core/operation/data_evolution_file_store_scan.h"
+#include <algorithm>
+#include <limits>
#include <map>
#include <string>
-#include "paimon/common/data/blob_utils.h"
#include "paimon/common/reader/data_evolution_array.h"
#include "paimon/common/reader/data_evolution_row.h"
#include "paimon/common/utils/object_utils.h"
#include "paimon/common/utils/range_helper.h"
+#include "paimon/core/utils/data_evolution_utils.h"
namespace paimon {
Result<bool> DataEvolutionFileStoreScan::FilterEntryByRowRanges(
const ManifestEntry& entry, const std::optional<RowRangeIndex>& row_range_index) {
@@ -114,19 +116,30 @@ Result<std::pair<int64_t, SimpleStatsEvolution::EvolutionStats>>
DataEvolutionFileStoreScan::EvolutionStats(
const std::vector<ManifestEntry>& old_entries, const std::shared_ptr<TableSchema>& table_schema,
const std::function<Result<std::shared_ptr<TableSchema>>(int64_t)>& schema_fetcher) {
- // exclude blob files, useless for predicate eval
+ if (old_entries.empty()) {
+ return Status::Invalid(
+ "DataEvolutionFileStoreScan EvolutionStats: entries cannot be empty");
+ }
+ // Independently rolled dedicated files cover only part of the logical group's row range.
+ int64_t group_start = std::numeric_limits<int64_t>::max();
+ int64_t group_end = 0;
std::vector<ManifestEntry> entries;
entries.reserve(old_entries.size());
for (const auto& entry : old_entries) {
- if (!BlobUtils::IsBlobFile(entry.File()->file_name)) {
+ const auto& file = entry.File();
+ PAIMON_ASSIGN_OR_RAISE(int64_t first_row_id, file->NonNullFirstRowId());
+ if (first_row_id < 0 || file->row_count <= 0 ||
+ first_row_id > std::numeric_limits<int64_t>::max() - file->row_count) {
+ return Status::Invalid("Invalid row id range in data-evolution statistics group");
+ }
+ group_start = std::min(group_start, first_row_id);
+ group_end = std::max(group_end, first_row_id + file->row_count);
+ // BLOB and VECTOR files do not contribute column statistics.
+ if (DataEvolutionUtils::IsNormalFile(file->file_name)) {
entries.push_back(entry);
}
}
- if (entries.empty()) {
- return Status::Invalid(
- "DataEvolutionFileStoreScan EvolutionStats: after exclude blob files, entries cannot "
- "be empty");
- }
+ const int64_t group_row_count = group_end - group_start;
std::stable_sort(entries.begin(), entries.end(),
[](const ManifestEntry& e1, const ManifestEntry& e2) {
return e1.File()->max_sequence_number > e2.File()->max_sequence_number;
@@ -166,9 +179,10 @@ DataEvolutionFileStoreScan::EvolutionStats(
// -2 indicates that the table fields will be matched to the current file.
row_offsets[table_field_idx] = -2;
auto stats_iter = stats_field_map.find(write_field.Name());
- // if the fields are of the same type and contain statistics, then update
- // row_offsets, otherwise evolved stats of current field is null
- if (table_field.Type()->Equals(write_field.Type()) &&
+ // Only full-range providers can supply statistics for the entire group.
+ if (file_meta->first_row_id == group_start &&
+ file_meta->row_count == group_row_count &&
+ table_field.Type()->Equals(write_field.Type()) &&
stats_iter != stats_field_map.end()) {
row_offsets[table_field_idx] = entry_idx;
field_offsets[table_field_idx] = stats_iter->second;
@@ -190,14 +204,20 @@ DataEvolutionFileStoreScan::EvolutionStats(
max_rows.push_back(stats.MaxValues());
null_counts.push_back(stats.NullCounts());
}
+ if (entries.empty()) {
+ // Dedicated-only incremental groups have unknown statistics for every column.
+ const auto& empty_stats = SimpleStats::EmptyStats();
+ min_rows.push_back(empty_stats.MinValues());
+ max_rows.push_back(empty_stats.MaxValues());
+ null_counts.push_back(empty_stats.NullCounts());
+ }
auto final_min = std::make_shared<DataEvolutionRow>(min_rows, row_offsets, field_offsets);
auto final_max = std::make_shared<DataEvolutionRow>(max_rows, row_offsets, field_offsets);
auto final_null_counts =
std::make_shared<DataEvolutionArray>(null_counts, row_offsets, field_offsets);
- return std::make_pair(
- entries[0].File()->row_count,
- SimpleStatsEvolution::EvolutionStats(final_min, final_max, final_null_counts));
+ return std::make_pair(group_row_count, SimpleStatsEvolution::EvolutionStats(
+ final_min, final_max, final_null_counts));
}
} // namespace paimon
diff --git a/src/paimon/core/operation/data_evolution_file_store_scan_test.cpp b/src/paimon/core/operation/data_evolution_file_store_scan_test.cpp
index 7d042814..623e91cd 100644
--- a/src/paimon/core/operation/data_evolution_file_store_scan_test.cpp
+++ b/src/paimon/core/operation/data_evolution_file_store_scan_test.cpp
@@ -20,19 +20,26 @@
#include <algorithm>
#include <cstdint>
+#include <map>
#include <string>
#include <utility>
#include <vector>
+#include "fmt/format.h"
#include "gtest/gtest.h"
#include "paimon/common/data/binary_row.h"
#include "paimon/common/data/binary_row_writer.h"
#include "paimon/common/data/blob_utils.h"
#include "paimon/common/reader/data_evolution_array.h"
#include "paimon/common/reader/data_evolution_row.h"
+#include "paimon/core/core_options.h"
#include "paimon/core/schema/table_schema.h"
#include "paimon/core/stats/simple_stats_evolution.h"
+#include "paimon/executor.h"
#include "paimon/memory/memory_pool.h"
+#include "paimon/predicate/literal.h"
+#include "paimon/predicate/predicate_builder.h"
+#include "paimon/scan_context.h"
#include "paimon/testing/utils/binary_row_generator.h"
#include "paimon/testing/utils/testharness.h"
@@ -55,16 +62,16 @@ class DataEvolutionFileStoreScanTest : public ::testing::Test {
ManifestEntry CreateManifestEntry(
int64_t schema_id, const SimpleStats& stats,
const std::optional<std::vector<std::string>>& value_stats_cols,
- const std::optional<std::vector<std::string>>& write_cols,
- int64_t sequence_number = 0) const {
+ const std::optional<std::vector<std::string>>& write_cols, int64_t sequence_number = 0,
+ const std::string& file_name = "test-file.parquet", int64_t row_count = 100,
+ int64_t first_row_id = 0) const {
EXPECT_OK_AND_ASSIGN(
auto meta,
- DataFileMeta::ForAppend(
- /*file_name=*/"test-file.parquet", /*file_size=*/100l, /*row_count=*/100l, stats,
- /*min_sequence_number=*/sequence_number,
- /*max_sequence_number=*/sequence_number, schema_id,
- /*file_source=*/FileSource::Append(), value_stats_cols,
- /*external_path=*/std::nullopt, /*first_row_id=*/0, write_cols));
+ DataFileMeta::ForAppend(file_name, /*file_size=*/100l, row_count, stats,
+ /*min_sequence_number=*/sequence_number,
+ /*max_sequence_number=*/sequence_number, schema_id,
+ /*file_source=*/FileSource::Append(), value_stats_cols,
+ /*external_path=*/std::nullopt, first_row_id, write_cols));
return ManifestEntry(FileKind::Add(), /*partition=*/BinaryRow::EmptyRow(), /*bucket=*/0,
/*total_buckets=*/1, meta);
}
@@ -543,6 +550,139 @@ TEST_F(DataEvolutionFileStoreScanTest, TestEvolutionStatsWithBlob) {
ASSERT_TRUE(null_counts->IsNullAt(2));
}
+TEST_F(DataEvolutionFileStoreScanTest, TestEvolutionStatsWithRolledVectorFiles) {
+ std::vector<DataField> fields = {
+ DataField(0, arrow::field("id", arrow::int32())),
+ DataField(1, arrow::field("embedding", arrow::fixed_size_list(arrow::float32(), 3))),
+ DataField(2, arrow::field("tag", arrow::utf8())),
+ };
+ auto table_schema = CreateTableSchema(/*schema_id=*/0, fields);
+ auto arrow_schema = DataField::ConvertDataFieldsToArrowSchema(fields);
+ auto schema_fetcher = [&](int64_t) -> Result<std::shared_ptr<TableSchema>> {
+ return table_schema;
+ };
+ constexpr int64_t kFirstRowId = 123;
+ constexpr int64_t kRowCount = 10000;
+ constexpr int64_t kVectorRows = 1000;
+ auto main_entry = CreateManifestEntry(
+ /*schema_id=*/0,
+ BinaryRowGenerator::GenerateStats({0, std::string("a")}, {9999, std::string("z")},
+ {0, 1000}, pool_.get()),
+ /*value_stats_cols=*/std::nullopt,
+ /*write_cols=*/std::vector<std::string>{"id", "tag"},
+ /*sequence_number=*/1, "main.parquet", kRowCount, kFirstRowId);
+ std::vector<ManifestEntry> entries = {main_entry};
+ for (int64_t i = 0; i < kRowCount / kVectorRows; ++i) {
+ entries.push_back(CreateManifestEntry(
+ /*schema_id=*/0, SimpleStats::EmptyStats(),
+ /*value_stats_cols=*/std::vector<std::string>{},
+ /*write_cols=*/std::vector<std::string>{"embedding"},
+ /*sequence_number=*/i + 2, fmt::format("data-{}.vector.vortex", i), kVectorRows,
+ kFirstRowId + i * kVectorRows));
+ }
+ // VECTOR updates are newer than the full-range main file and cover only 1,000 rows each.
+ for (bool reverse_order : {false, true}) {
+ SCOPED_TRACE(reverse_order);
+ if (reverse_order) {
+ std::reverse(entries.begin(), entries.end());
+ }
+ ASSERT_OK_AND_ASSIGN(auto result_stats, DataEvolutionFileStoreScan::EvolutionStats(
+ entries, table_schema, schema_fetcher));
+ ASSERT_EQ(result_stats.first, kRowCount);
+ const auto& stats = result_stats.second;
+ ASSERT_EQ(stats.min_values->GetString(2).ToString(), "a");
+ ASSERT_EQ(stats.max_values->GetString(2).ToString(), "z");
+ ASSERT_EQ(stats.null_counts->GetLong(2), kVectorRows);
+ ASSERT_TRUE(stats.min_values->IsNullAt(1));
+ ASSERT_TRUE(stats.max_values->IsNullAt(1));
+ ASSERT_TRUE(stats.null_counts->IsNullAt(1));
+
+ const std::vector<std::pair<std::shared_ptr<Predicate>, bool>> cases = {
+ {PredicateBuilder::IsNotNull(2, "tag", FieldType::STRING), true},
+ {PredicateBuilder::IsNull(2, "tag", FieldType::STRING), true},
+ {PredicateBuilder::IsNull(0, "id", FieldType::INT), false},
+ {PredicateBuilder::Equal(0, "id", FieldType::INT, Literal(10000)), false},
+ };
+ ASSERT_OK_AND_ASSIGN(CoreOptions options, CoreOptions::FromMap({}));
+ for (const auto& [predicate, keep] : cases) {
+ SCOPED_TRACE(predicate->ToString());
+ auto filters = std::make_shared<ScanFilter>(
+ predicate, std::vector<std::map<std::string, std::string>>{}, std::nullopt);
+ ASSERT_OK_AND_ASSIGN(auto scan,
+ DataEvolutionFileStoreScan::Create(
+ nullptr, nullptr, nullptr, nullptr, table_schema, arrow_schema,
+ filters, options, CreateDefaultExecutor(), pool_));
+ ASSERT_OK_AND_ASSIGN(auto filtered, scan->PostFilterManifestEntries(
+ std::vector<ManifestEntry>(entries)));
+ ASSERT_EQ(filtered.size(), keep ? entries.size() : 0);
+ }
+ }
+}
+
+TEST_F(DataEvolutionFileStoreScanTest, TestEvolutionStatsWithOnlyVectorFiles) {
+ std::vector<DataField> fields = {
+ DataField(0, arrow::field("embedding", arrow::fixed_size_list(arrow::float32(), 3))),
+ };
+ auto table_schema = CreateTableSchema(/*schema_id=*/0, fields);
+ auto schema_fetcher = [&](int64_t) -> Result<std::shared_ptr<TableSchema>> {
+ ADD_FAILURE() << "Dedicated VECTOR files must not contribute column statistics";
+ return table_schema;
+ };
+ // An incremental scan can contain only a contiguous subset of independently rolled files.
+ auto first = CreateManifestEntry(
+ /*schema_id=*/0, SimpleStats::EmptyStats(),
+ /*value_stats_cols=*/std::vector<std::string>{},
+ /*write_cols=*/std::vector<std::string>{"embedding"},
+ /*sequence_number=*/2, "first.vector.vortex", /*row_count=*/1000, /*first_row_id=*/123);
+ auto second = CreateManifestEntry(
+ /*schema_id=*/0, SimpleStats::EmptyStats(),
+ /*value_stats_cols=*/std::vector<std::string>{},
+ /*write_cols=*/std::vector<std::string>{"embedding"},
+ /*sequence_number=*/3, "second.vector.vortex", /*row_count=*/500,
+ /*first_row_id=*/1123);
+ ASSERT_OK_AND_ASSIGN(auto result_stats, DataEvolutionFileStoreScan::EvolutionStats(
+ {second, first}, table_schema, schema_fetcher));
+ ASSERT_EQ(result_stats.first, 1500);
+ ASSERT_TRUE(result_stats.second.min_values->IsNullAt(0));
+ ASSERT_TRUE(result_stats.second.max_values->IsNullAt(0));
+ ASSERT_TRUE(result_stats.second.null_counts->IsNullAt(0));
+}
+
+TEST_F(DataEvolutionFileStoreScanTest, TestEvolutionStatsWithPartialRangeProviders) {
+ std::vector<DataField> fields = {
+ DataField(0, arrow::field("tag", arrow::utf8())),
+ DataField(1, arrow::field("embedding", arrow::fixed_size_list(arrow::float32(), 3))),
+ };
+ auto table_schema = CreateTableSchema(/*schema_id=*/0, fields);
+ auto schema_fetcher = [&](int64_t) -> Result<std::shared_ptr<TableSchema>> {
+ return table_schema;
+ };
+ auto first = CreateManifestEntry(
+ /*schema_id=*/0,
+ BinaryRowGenerator::GenerateStats({std::string("a")}, {std::string("z")}, {0}, pool_.get()),
+ /*value_stats_cols=*/std::nullopt, /*write_cols=*/std::vector<std::string>{"tag"},
+ /*sequence_number=*/1, "first.parquet", /*row_count=*/1000, /*first_row_id=*/123);
+ auto second = CreateManifestEntry(
+ /*schema_id=*/0,
+ BinaryRowGenerator::GenerateStats({std::string("a")}, {std::string("z")}, {500},
+ pool_.get()),
+ /*value_stats_cols=*/std::nullopt, /*write_cols=*/std::vector<std::string>{"tag"},
+ /*sequence_number=*/2, "second.parquet", /*row_count=*/1000, /*first_row_id=*/1123);
+ auto vector = CreateManifestEntry(
+ /*schema_id=*/0, SimpleStats::EmptyStats(),
+ /*value_stats_cols=*/std::vector<std::string>{},
+ /*write_cols=*/std::vector<std::string>{"embedding"},
+ /*sequence_number=*/3, "data.vector.vortex", /*row_count=*/2000, /*first_row_id=*/123);
+ ASSERT_OK_AND_ASSIGN(auto result_stats,
+ DataEvolutionFileStoreScan::EvolutionStats({first, vector, second},
+ table_schema, schema_fetcher));
+ ASSERT_EQ(result_stats.first, 2000);
+ // The VECTOR range joins two main files; neither file's stats describe the whole group.
+ ASSERT_TRUE(result_stats.second.min_values->IsNullAt(0));
+ ASSERT_TRUE(result_stats.second.max_values->IsNullAt(0));
+ ASSERT_TRUE(result_stats.second.null_counts->IsNullAt(0));
+}
+
TEST_F(DataEvolutionFileStoreScanTest, TestFilterEntryByRowRanges) {
// row id range [100, 200)
auto file = std::make_shared<DataFileMeta>(
diff --git a/src/paimon/core/operation/data_evolution_split_read.cpp b/src/paimon/core/operation/data_evolution_split_read.cpp
index 7491ad3e..e3a52c6f 100644
--- a/src/paimon/core/operation/data_evolution_split_read.cpp
+++ b/src/paimon/core/operation/data_evolution_split_read.cpp
@@ -51,11 +51,13 @@
#include "paimon/common/utils/object_utils.h"
#include "paimon/common/utils/path_util.h"
#include "paimon/common/utils/range_helper.h"
+#include "paimon/common/utils/vector_store_utils.h"
#include "paimon/core/core_options.h"
#include "paimon/core/deletionvectors/apply_deletion_vector_batch_reader.h"
#include "paimon/core/global_index/indexed_split_impl.h"
#include "paimon/core/io/file_index_evaluator.h"
#include "paimon/core/io/file_index_options.h"
+#include "paimon/core/table/format/lazy_concat_batch_reader.h"
#include "paimon/core/utils/blob_view_lookup.h"
#include "paimon/core/utils/data_evolution_utils.h"
#include "paimon/core/utils/field_mapping.h"
@@ -345,11 +347,11 @@ Result<std::unique_ptr<BatchReader>> DataEvolutionSplitRead::CreateBlobViewReade
std::vector<std::unique_ptr<BatchReader>> batch_readers;
for (const std::vector<std::shared_ptr<DataFileMeta>>& group : split_by_row_id) {
- // skip blob files: they only contain blob payloads, not the blob-view columns.
+ // Dedicated files contain payloads, not the blob-view columns.
std::vector<std::shared_ptr<DataFileMeta>> data_files;
data_files.reserve(group.size());
for (const auto& file : group) {
- if (!BlobUtils::IsBlobFile(file->file_name)) {
+ if (DataEvolutionUtils::IsNormalFile(file->file_name)) {
data_files.push_back(file);
}
}
@@ -475,7 +477,7 @@ Result<std::unique_ptr<BatchReader>> DataEvolutionSplitRead::InnerCreateReader(
sub_readers.push_back(std::move(raw_file_readers[0]));
} else {
PAIMON_ASSIGN_OR_RAISE(
- std::unique_ptr<DataEvolutionFileReader> evolution_reader,
+ std::unique_ptr<BatchReader> evolution_reader,
CreateUnionReader(split_impl->Partition(), need_merge_files, row_ranges,
data_file_path_factory, group_dv_factory, group_dv, read_schema));
sub_readers.push_back(std::move(evolution_reader));
@@ -674,7 +676,203 @@ DataEvolutionSplitRead::SplitFieldBunches(
return fields_files;
}
-Result<std::unique_ptr<DataEvolutionFileReader>> DataEvolutionSplitRead::CreateUnionReader(
+Result<std::vector<DataEvolutionSplitRead::VectorReadRange>>
+DataEvolutionSplitRead::PlanVectorReadRanges(
+ const std::vector<std::shared_ptr<DataFileMeta>>& files,
+ const std::vector<int32_t>& read_field_ids,
+ const std::function<Result<std::vector<int32_t>>(const std::shared_ptr<DataFileMeta>&)>&
+ file_to_field_ids) {
+ struct Candidate {
+ std::shared_ptr<DataFileMeta> file;
+ std::vector<int32_t> field_ids;
+ };
+ std::vector<Candidate> candidates;
+ std::vector<std::shared_ptr<DataFileMeta>> other_files;
+ std::map<int64_t, std::vector<size_t>> boundaries;
+ std::unordered_set<int32_t> projected_ids(read_field_ids.begin(), read_field_ids.end());
+ std::optional<Range> logical_range;
+ int64_t first_row_id = std::numeric_limits<int64_t>::max();
+ int64_t end_row_id = std::numeric_limits<int64_t>::min();
+ bool vector_only = std::all_of(files.begin(), files.end(), [](const auto& file) {
+ return VectorStoreUtils::IsVectorStoreFile(file->file_name);
+ });
+ for (const auto& file : files) {
+ PAIMON_ASSIGN_OR_RAISE(int64_t first, file->NonNullFirstRowId());
+ if (first < 0 || file->row_count <= 0 ||
+ first > std::numeric_limits<int64_t>::max() - file->row_count) {
+ return Status::Invalid("Invalid row id range in VECTOR merge group");
+ }
+ int64_t end = first + file->row_count;
+ first_row_id = std::min(first_row_id, first);
+ end_row_id = std::max(end_row_id, end);
+ if (!VectorStoreUtils::IsVectorStoreFile(file->file_name)) {
+ other_files.push_back(file);
+ if (DataEvolutionUtils::IsNormalFile(file->file_name)) {
+ logical_range = Range(first, end - 1);
+ }
+ continue;
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::vector<int32_t> field_ids, file_to_field_ids(file));
+ field_ids.erase(std::remove_if(field_ids.begin(), field_ids.end(),
+ [&](int32_t id) { return projected_ids.count(id) == 0; }),
+ field_ids.end());
+ if (!field_ids.empty() || vector_only) {
+ size_t index = candidates.size();
+ candidates.push_back(Candidate{file, std::move(field_ids)});
+ boundaries[first].push_back(index);
+ boundaries[end].push_back(index);
+ }
+ }
+ if (files.empty()) {
+ return std::vector<VectorReadRange>();
+ }
+ if (!logical_range) {
+ logical_range = Range(first_row_id, end_row_id - 1);
+ }
+ boundaries[logical_range->from];
+ boundaries[logical_range->to + 1];
+
+ auto newest_first = [&](size_t left, size_t right) {
+ const auto& lhs = candidates[left].file;
+ const auto& rhs = candidates[right].file;
+ if (lhs->max_sequence_number != rhs->max_sequence_number) {
+ return lhs->max_sequence_number > rhs->max_sequence_number;
+ }
+ if (lhs->file_name != rhs->file_name) {
+ return lhs->file_name < rhs->file_name;
+ }
+ return left < right;
+ };
+ std::set<size_t, decltype(newest_first)> active(newest_first);
+ std::vector<VectorReadRange> result;
+ int64_t start = boundaries.begin()->first;
+ for (const auto& [end, events] : boundaries) {
+ int64_t from = std::max(start, logical_range->from);
+ int64_t to = std::min(end - 1, logical_range->to);
+ if (from <= to) {
+ std::unordered_set<int32_t> assigned;
+ std::vector<std::shared_ptr<DataFileMeta>> providers = other_files;
+ for (size_t index : active) {
+ const auto& candidate = candidates[index];
+ bool provides_field = false;
+ for (int32_t id : candidate.field_ids) {
+ provides_field |= assigned.insert(id).second;
+ }
+ // An explicit NULL in a newer file overrides the older value as well.
+ if (provides_field) {
+ providers.push_back(candidate.file);
+ }
+ }
+ if (providers.empty() && vector_only && !active.empty()) {
+ providers.push_back(candidates[*active.begin()].file);
+ }
+ // Ordinary inline fields and dedicated fields share the same sequence precedence.
+ // BLOB files remain last and retain their own per-row fallback ordering.
+ std::stable_sort(
+ providers.begin(), providers.end(), [](const auto& lhs, const auto& rhs) {
+ bool lhs_blob = BlobUtils::IsBlobFile(lhs->file_name);
+ bool rhs_blob = BlobUtils::IsBlobFile(rhs->file_name);
+ if (lhs_blob != rhs_blob) {
+ return !lhs_blob;
+ }
+ return !lhs_blob && lhs->max_sequence_number > rhs->max_sequence_number;
+ });
+ if (!result.empty() && result.back().range.to + 1 == from &&
+ result.back().files == providers) {
+ result.back().range.to = to;
+ } else {
+ result.push_back(VectorReadRange{Range(from, to), std::move(providers)});
+ }
+ }
+ for (size_t index : events) {
+ if (active.erase(index) == 0) {
+ active.insert(index);
+ }
+ }
+ start = end;
+ }
+ return result;
+}
+
+Result<std::unique_ptr<BatchReader>> DataEvolutionSplitRead::CreateUnionReader(
+ const BinaryRow& partition, const std::vector<std::shared_ptr<DataFileMeta>>& need_merge_files,
+ const std::optional<std::vector<Range>>& row_ranges,
+ const std::shared_ptr<DataFilePathFactory>& data_file_path_factory,
+ const DeletionVector::Factory& group_dv_factory,
+ const std::optional<GroupDeletionVector>& group_dv,
+ const std::shared_ptr<arrow::Schema>& read_schema) const {
+ if (std::none_of(need_merge_files.begin(), need_merge_files.end(), [](const auto& file) {
+ return VectorStoreUtils::IsVectorStoreFile(file->file_name);
+ })) {
+ return CreateFieldUnionReader(partition, need_merge_files, row_ranges,
+ data_file_path_factory, group_dv_factory, group_dv,
+ read_schema);
+ }
+
+ PAIMON_ASSIGN_OR_RAISE(std::vector<DataField> read_fields,
+ DataField::ConvertArrowSchemaToDataFields(read_schema));
+ std::vector<int32_t> read_field_ids;
+ for (const auto& field : read_fields) {
+ if (!SpecialFields::IsSystemField(field.Name())) {
+ read_field_ids.push_back(field.Id());
+ }
+ }
+ auto file_to_field_ids =
+ [&](const std::shared_ptr<DataFileMeta>& file) -> Result<std::vector<int32_t>> {
+ if (!file->write_cols) {
+ return Status::Invalid("VECTOR files must declare their write columns");
+ }
+ auto data_schema = context_->GetTableSchema();
+ if (file->schema_id != data_schema->Id()) {
+ PAIMON_ASSIGN_OR_RAISE(data_schema, schema_manager_->ReadSchema(file->schema_id));
+ }
+ std::vector<int32_t> ids;
+ for (const auto& name : file->write_cols.value()) {
+ if (SpecialFields::IsSystemField(name)) {
+ continue;
+ }
+ PAIMON_ASSIGN_OR_RAISE(DataField field, data_schema->GetField(name));
+ ids.push_back(field.Id());
+ }
+ return ids;
+ };
+ PAIMON_ASSIGN_OR_RAISE(
+ std::vector<VectorReadRange> ranges,
+ PlanVectorReadRanges(need_merge_files, read_field_ids, file_to_field_ids));
+ // Each range creates a separate field-union reader, reopening a main file that spans ranges.
+ // TODO(ChaomingZhangCN): Keep one main reader and concatenate VECTOR readers for contiguous,
+ // non-overlapping files whose field providers can be merged sequentially.
+ std::vector<LazyConcatBatchReader::Source> sources;
+ auto split_read =
+ std::make_shared<DataEvolutionSplitRead>(path_factory_, context_, pool_, executor_);
+ std::optional<std::vector<Range>> normalized_ranges;
+ if (row_ranges) {
+ normalized_ranges = Range::SortAndMergeOverlap(row_ranges.value(), /*adjacent=*/true);
+ }
+ for (const auto& range : ranges) {
+ std::vector<Range> selected = {range.range};
+ if (normalized_ranges) {
+ selected = Range::And(selected, normalized_ranges.value());
+ }
+ if (selected.empty() || range.files.empty()) {
+ continue;
+ }
+ // Keep original file metadata and the group's DV; selection uses absolute row ids.
+ // Open one range at a time, so independently rolled files do not multiply open readers.
+ sources.push_back(LazyConcatBatchReader::Source{
+ range.range.ToString(),
+ [split_read, partition, files = range.files, selected = std::move(selected),
+ data_file_path_factory, group_dv_factory, group_dv,
+ read_schema]() -> Result<std::unique_ptr<BatchReader>> {
+ return split_read->CreateFieldUnionReader(partition, files, selected,
+ data_file_path_factory, group_dv_factory,
+ group_dv, read_schema);
+ }});
+ }
+ return std::make_unique<LazyConcatBatchReader>(std::move(sources), arrow_pool_);
+}
+
+Result<std::unique_ptr<BatchReader>> DataEvolutionSplitRead::CreateFieldUnionReader(
const BinaryRow& partition, const std::vector<std::shared_ptr<DataFileMeta>>& need_merge_files,
const std::optional<std::vector<Range>>& row_ranges,
const std::shared_ptr<DataFilePathFactory>& data_file_path_factory,
@@ -796,10 +994,26 @@ Result<std::unique_ptr<DataEvolutionFileReader>> DataEvolutionSplitRead::CreateU
}
}
}
+ if (std::none_of(file_batch_readers.begin(), file_batch_readers.end(),
+ [](const auto& reader) { return reader != nullptr; })) {
+ // A projected VECTOR can be absent in this range. Let schema evolution fill NULLs,
+ // using an original file as the row-count provider even when no field is projected.
+ PAIMON_ASSIGN_OR_RAISE(
+ std::vector<std::unique_ptr<FileBatchReader>> file_readers,
+ CreateRawFileReaders(partition, fields_files[0]->Files(), read_schema,
+ /*predicate=*/nullptr, group_dv_factory, row_ranges,
+ data_file_path_factory, /*extra_format_options=*/{}));
+ auto readers =
+ ObjectUtils::MoveVector<std::unique_ptr<BatchReader>>(std::move(file_readers));
+ return std::make_unique<ConcatBatchReader>(std::move(readers), arrow_pool_);
+ }
// TODO(xinyu.lxy): check nullable when reader_offsets[read_field_idx] = -1
- return DataEvolutionFileReader::Create(std::move(file_batch_readers), read_schema,
- options_.GetReadBatchSize(), reader_offsets,
- field_offsets, arrow_pool_);
+ PAIMON_ASSIGN_OR_RAISE(
+ std::unique_ptr<BatchReader> reader,
+ DataEvolutionFileReader::Create(std::move(file_batch_readers), read_schema,
+ options_.GetReadBatchSize(), reader_offsets, field_offsets,
+ arrow_pool_));
+ return reader;
}
namespace {
@@ -876,15 +1090,19 @@ Result<std::unique_ptr<BatchReader>> DataEvolutionSplitRead::CreateBlobFallbackR
BlobFallbackBatchReader::Segment{nullptr, std::move(gap_selected_ranges)});
}
}
- PAIMON_ASSIGN_OR_RAISE(
- std::vector<std::unique_ptr<FileBatchReader>> file_readers,
- CreateRawFileReaders(partition, {file}, file_read_schema,
- /*predicate=*/nullptr, group_dv_factory, row_ranges,
- data_file_path_factory, blob_format_options));
- if (file_readers.size() != 1) {
- return Status::Invalid("Unexpected: blob fallback file reader was skipped.");
+ if (!SelectedRangesInRange(first_row_id, first_row_id + file->row_count - 1, row_ranges)
+ .empty()) {
+ PAIMON_ASSIGN_OR_RAISE(
+ std::vector<std::unique_ptr<FileBatchReader>> file_readers,
+ CreateRawFileReaders(partition, {file}, file_read_schema,
+ /*predicate=*/nullptr, group_dv_factory, row_ranges,
+ data_file_path_factory, blob_format_options));
+ if (file_readers.size() != 1) {
+ return Status::Invalid("Unexpected: blob fallback file reader was skipped.");
+ }
+ segments.push_back(
+ BlobFallbackBatchReader::Segment{std::move(file_readers[0]), {}});
}
- segments.push_back(BlobFallbackBatchReader::Segment{std::move(file_readers[0]), {}});
next_row_id = first_row_id + file->row_count;
}
if (next_row_id <= union_last_row_id) {
@@ -927,16 +1145,20 @@ DataEvolutionSplitRead::MergeRangesAndSort(std::vector<std::shared_ptr<DataFileM
PAIMON_ASSIGN_OR_RAISE(std::vector<std::vector<std::shared_ptr<DataFileMeta>>> result,
range_helper.MergeOverlappingRanges(std::move(files)));
- // in group, sort by blob file and max_seq
+ // Keep ordinary files separate from independently rolled dedicated files.
for (auto& group : result) {
// split to data files and blob files
std::vector<std::shared_ptr<DataFileMeta>> data_files;
data_files.reserve(group.size());
std::vector<std::shared_ptr<DataFileMeta>> blob_files;
blob_files.reserve(group.size());
+ std::vector<std::shared_ptr<DataFileMeta>> vector_files;
+ vector_files.reserve(group.size());
for (auto& meta : group) {
if (BlobUtils::IsBlobFile(meta->file_name)) {
blob_files.push_back(std::move(meta));
+ } else if (VectorStoreUtils::IsVectorStoreFile(meta->file_name)) {
+ vector_files.push_back(std::move(meta));
} else {
data_files.push_back(std::move(meta));
}
@@ -963,10 +1185,18 @@ DataEvolutionSplitRead::MergeRangesAndSort(std::vector<std::shared_ptr<DataFileM
}
return m1->max_sequence_number > m2->max_sequence_number;
});
- // concat data files and blob files
+ std::stable_sort(vector_files.begin(), vector_files.end(),
+ [](const auto& lhs, const auto& rhs) {
+ if (lhs->first_row_id != rhs->first_row_id) {
+ return lhs->first_row_id < rhs->first_row_id;
+ }
+ return lhs->max_sequence_number > rhs->max_sequence_number;
+ });
group.clear();
group.insert(group.end(), std::make_move_iterator(data_files.begin()),
std::make_move_iterator(data_files.end()));
+ group.insert(group.end(), std::make_move_iterator(vector_files.begin()),
+ std::make_move_iterator(vector_files.end()));
group.insert(group.end(), std::make_move_iterator(blob_files.begin()),
std::make_move_iterator(blob_files.end()));
}
diff --git a/src/paimon/core/operation/data_evolution_split_read.h b/src/paimon/core/operation/data_evolution_split_read.h
index 2c0eabe0..b51dd447 100644
--- a/src/paimon/core/operation/data_evolution_split_read.h
+++ b/src/paimon/core/operation/data_evolution_split_read.h
@@ -215,6 +215,18 @@ class DataEvolutionSplitRead : public AbstractSplitRead {
static Result<std::vector<Range>> ExcludeDeletedRowIds(const std::vector<Range>& ranges,
const GroupDeletionVector& group_dv);
+ struct VectorReadRange {
+ Range range;
+ std::vector<std::shared_ptr<DataFileMeta>> files;
+ };
+
+ /// Split at VECTOR file boundaries so each field has one latest provider per range.
+ static Result<std::vector<VectorReadRange>> PlanVectorReadRanges(
+ const std::vector<std::shared_ptr<DataFileMeta>>& files,
+ const std::vector<int32_t>& read_field_ids,
+ const std::function<Result<std::vector<int32_t>>(const std::shared_ptr<DataFileMeta>&)>&
+ file_to_field_ids);
+
static Result<std::vector<std::shared_ptr<DataEvolutionSplitRead::FieldBunch>>>
SplitFieldBunches(const std::vector<std::shared_ptr<DataFileMeta>>& need_merge_files,
const std::function<Result<int32_t>(const std::shared_ptr<DataFileMeta>&)>&
@@ -248,7 +260,16 @@ class DataEvolutionSplitRead : public AbstractSplitRead {
const std::optional<std::vector<Range>>& row_ranges) const;
private:
- Result<std::unique_ptr<DataEvolutionFileReader>> CreateUnionReader(
+ Result<std::unique_ptr<BatchReader>> CreateUnionReader(
+ const BinaryRow& partition,
+ const std::vector<std::shared_ptr<DataFileMeta>>& need_merge_files,
+ const std::optional<std::vector<Range>>& row_ranges,
+ const std::shared_ptr<DataFilePathFactory>& data_file_path_factory,
+ const DeletionVector::Factory& group_dv_factory,
+ const std::optional<GroupDeletionVector>& group_dv,
+ const std::shared_ptr<arrow::Schema>& read_schema) const;
+
+ Result<std::unique_ptr<BatchReader>> CreateFieldUnionReader(
const BinaryRow& partition,
const std::vector<std::shared_ptr<DataFileMeta>>& need_merge_files,
const std::optional<std::vector<Range>>& row_ranges,
diff --git a/src/paimon/core/operation/data_evolution_split_read_test.cpp b/src/paimon/core/operation/data_evolution_split_read_test.cpp
index 39fdb100..0b113725 100644
--- a/src/paimon/core/operation/data_evolution_split_read_test.cpp
+++ b/src/paimon/core/operation/data_evolution_split_read_test.cpp
@@ -18,6 +18,7 @@
#include "paimon/core/operation/data_evolution_split_read.h"
+#include <algorithm>
#include <cstdint>
#include <map>
#include <optional>
@@ -26,6 +27,7 @@
#include "gtest/gtest.h"
#include "paimon/common/data/binary_row.h"
#include "paimon/common/table/special_fields.h"
+#include "paimon/common/utils/vector_store_utils.h"
#include "paimon/core/deletionvectors/bitmap64_deletion_vector.h"
#include "paimon/core/deletionvectors/bitmap_deletion_vector.h"
#include "paimon/core/io/data_file_meta.h"
@@ -102,6 +104,40 @@ class DataEvolutionSplitReadTest : public ::testing::Test {
.value();
}
+ static Result<std::vector<int32_t>> EmbeddingFieldIds(
+ const std::shared_ptr<DataFileMeta>& file) {
+ if (!VectorStoreUtils::IsVectorStoreFile(file->file_name)) {
+ return Status::Invalid("Only VECTOR files need their field ids resolved");
+ }
+ return std::vector<int32_t>{1};
+ }
+
+ void CheckVectorReadProviders(
+ const std::vector<DataEvolutionSplitRead::VectorReadRange>& ranges, int64_t first_row_id,
+ const std::vector<std::vector<std::shared_ptr<DataFileMeta>>>& expected) const {
+ int64_t next_row_id = first_row_id;
+ for (const auto& range : ranges) {
+ ASSERT_EQ(range.range.from, next_row_id);
+ ASSERT_GE(range.range.to, range.range.from);
+ for (int64_t row_id = range.range.from; row_id <= range.range.to; ++row_id) {
+ SCOPED_TRACE(row_id);
+ const int64_t index = row_id - first_row_id;
+ ASSERT_LT(index, expected.size());
+ ASSERT_EQ(range.files.size(), expected[index].size());
+ for (const auto& file : expected[index]) {
+ ASSERT_NE(std::find(range.files.begin(), range.files.end(), file),
+ range.files.end());
+ }
+ for (size_t i = 1; i < range.files.size(); ++i) {
+ ASSERT_GE(range.files[i - 1]->max_sequence_number,
+ range.files[i]->max_sequence_number);
+ }
+ }
+ next_row_id = range.range.to + 1;
+ }
+ ASSERT_EQ(next_row_id, first_row_id + expected.size());
+ }
+
private:
std::shared_ptr<MemoryPool> pool_ = GetDefaultPool();
};
@@ -655,6 +691,188 @@ TEST_F(DataEvolutionSplitReadTest, TestDifferentRowIdRange) {
"Data files should be all row id ranges same.");
}
+TEST_F(DataEvolutionSplitReadTest, TestDifferentNormalRowIdRangeWithVectorFiles) {
+ std::vector<std::shared_ptr<DataFileMeta>> files = {
+ CreateNormalFile("file0.parquet", /*first_row_id=*/100, /*row_count=*/100,
+ /*max_sequence_number=*/10),
+ CreateNormalFile("file1.parquet", /*first_row_id=*/100, /*row_count=*/50,
+ /*max_sequence_number=*/20),
+ CreateDataFileMeta("file2.vector.vortex", /*first_row_id=*/100, /*row_count=*/40,
+ /*max_seq=*/10),
+ CreateDataFileMeta("file3.vector.vortex", /*first_row_id=*/140, /*row_count=*/60,
+ /*max_seq=*/10)};
+ ASSERT_NOK_WITH_MSG(DataEvolutionSplitRead::MergeRangesAndSort(std::move(files)),
+ "Data files should be all row id ranges same.");
+}
+
+TEST_F(DataEvolutionSplitReadTest, TestSplitWithMultipleVectorFilesPerGroup) {
+ auto main0 = CreateNormalFile("main0.parquet", /*first_row_id=*/100, /*row_count=*/100,
+ /*max_sequence_number=*/10);
+ auto vector0 = CreateDataFileMeta("vector0.vector.vortex", /*first_row_id=*/100,
+ /*row_count=*/40, /*max_seq=*/10);
+ auto vector1 = CreateDataFileMeta("vector1.vector.vortex", /*first_row_id=*/140,
+ /*row_count=*/60, /*max_seq=*/10);
+ auto main1 = CreateNormalFile("main1.parquet", /*first_row_id=*/200, /*row_count=*/100,
+ /*max_sequence_number=*/20);
+ auto vector2 = CreateDataFileMeta("vector2.vector.vortex", /*first_row_id=*/200,
+ /*row_count=*/80, /*max_seq=*/20);
+ auto vector3 = CreateDataFileMeta("vector3.vector.vortex", /*first_row_id=*/280,
+ /*row_count=*/20, /*max_seq=*/20);
+ std::vector<std::shared_ptr<DataFileMeta>> files = {vector1, main1, vector3,
+ vector0, main0, vector2};
+ ASSERT_OK_AND_ASSIGN(std::vector<std::vector<std::shared_ptr<DataFileMeta>>> split_metas,
+ DataEvolutionSplitRead::MergeRangesAndSort(std::move(files)));
+ std::vector<std::vector<std::shared_ptr<DataFileMeta>>> expected_metas = {
+ {main0, vector0, vector1}, {main1, vector2, vector3}};
+ ASSERT_EQ(expected_metas, split_metas);
+}
+
+TEST_F(DataEvolutionSplitReadTest, TestPlanIndependentlyRolledVectorReadRanges) {
+ auto main = CreateNormalFile("main.parquet", 100, 10, 0);
+ auto vector0 = CreateDataFileMeta("vector0.vector.vortex", 100, 3, 1);
+ auto vector1 = CreateDataFileMeta("vector1.vector.vortex", 103, 4, 1);
+ auto vector2 = CreateDataFileMeta("vector2.vector.vortex", 107, 3, 1);
+ ASSERT_OK_AND_ASSIGN(std::vector<DataEvolutionSplitRead::VectorReadRange> ranges,
+ DataEvolutionSplitRead::PlanVectorReadRanges(
+ {vector2, main, vector0, vector1}, {0, 1}, EmbeddingFieldIds));
+ ASSERT_EQ(ranges.size(), 3);
+ std::vector<std::vector<std::shared_ptr<DataFileMeta>>> expected(10, {main, vector1});
+ for (int32_t i : {0, 1, 2}) {
+ expected[i] = {main, vector0};
+ }
+ for (int32_t i : {7, 8, 9}) {
+ expected[i] = {main, vector2};
+ }
+ CheckVectorReadProviders(ranges, 100, expected);
+ // Planning must retain whole-file metadata so file-local row offsets and DVs remain valid.
+ ASSERT_EQ(main->first_row_id, 100);
+ ASSERT_EQ(main->row_count, 10);
+ ASSERT_EQ(vector1->first_row_id, 103);
+ ASSERT_EQ(vector1->row_count, 4);
+}
+
+TEST_F(DataEvolutionSplitReadTest, TestPlanVectorReadRangesPreservesOldPrefixAndSuffix) {
+ auto main = CreateNormalFile("main.parquet", 0, 10, 0);
+ auto old_vector = CreateDataFileMeta("old.vector.vortex", 0, 10, 1);
+ auto new_vector = CreateDataFileMeta("new.vector.vortex", 3, 3, 2);
+ ASSERT_OK_AND_ASSIGN(std::vector<DataEvolutionSplitRead::VectorReadRange> ranges,
+ DataEvolutionSplitRead::PlanVectorReadRanges(
+ {main, old_vector, new_vector}, {1}, EmbeddingFieldIds));
+ std::vector<std::vector<std::shared_ptr<DataFileMeta>>> expected(10, {main, old_vector});
+ for (int32_t i : {3, 4, 5}) {
+ expected[i] = {main, new_vector};
+ }
+ CheckVectorReadProviders(ranges, 0, expected);
+ ASSERT_EQ(old_vector->first_row_id, 0);
+ ASSERT_EQ(old_vector->row_count, 10);
+ ASSERT_EQ(new_vector->first_row_id, 3);
+ ASSERT_EQ(new_vector->row_count, 3);
+}
+
+TEST_F(DataEvolutionSplitReadTest, TestPlanVectorReadRangesWithInterleavedSequences) {
+ auto main = CreateNormalFile("main.parquet", 0, 10, 0);
+ auto vector0 = CreateDataFileMeta("vector0.vector.vortex", 0, 5, 4);
+ auto vector1 = CreateDataFileMeta("vector1.vector.vortex", 2, 6, 3);
+ auto vector2 = CreateDataFileMeta("vector2.vector.vortex", 5, 5, 2);
+ ASSERT_OK_AND_ASSIGN(std::vector<DataEvolutionSplitRead::VectorReadRange> ranges,
+ DataEvolutionSplitRead::PlanVectorReadRanges(
+ {vector2, vector1, main, vector0}, {1}, EmbeddingFieldIds));
+ std::vector<std::vector<std::shared_ptr<DataFileMeta>>> expected(10, {main, vector0});
+ for (int32_t i : {5, 6, 7}) {
+ expected[i] = {main, vector1};
+ }
+ for (int32_t i : {8, 9}) {
+ expected[i] = {main, vector2};
+ }
+ CheckVectorReadProviders(ranges, 0, expected);
+}
+
+TEST_F(DataEvolutionSplitReadTest, TestPlanVectorReadRangesKeepsProvidersForOtherFields) {
+ auto main = CreateNormalFile("main.parquet", 0, 10, 0);
+ auto old_vector = CreateDataFileMeta("old.vector.vortex", 0, 10, 1);
+ auto new_vector = CreateDataFileMeta("new.vector.vortex", 3, 3, 2);
+ auto file_to_field_ids =
+ [&old_vector](const std::shared_ptr<DataFileMeta>& file) -> Result<std::vector<int32_t>> {
+ return file == old_vector ? std::vector<int32_t>{1, 2} : std::vector<int32_t>{1};
+ };
+ ASSERT_OK_AND_ASSIGN(std::vector<DataEvolutionSplitRead::VectorReadRange> ranges,
+ DataEvolutionSplitRead::PlanVectorReadRanges(
+ {main, old_vector, new_vector}, {1, 2}, file_to_field_ids));
+ std::vector<std::vector<std::shared_ptr<DataFileMeta>>> expected(10, {main, old_vector});
+ for (int32_t i : {3, 4, 5}) {
+ expected[i] = {main, old_vector, new_vector};
+ }
+ CheckVectorReadProviders(ranges, 0, expected);
+}
+
+TEST_F(DataEvolutionSplitReadTest, TestPlanVectorReadRangesKeepsNormalFilesForGaps) {
+ auto main = CreateNormalFile("main.parquet", 0, 10, 0);
+ auto vector = CreateDataFileMeta("middle.vector.vortex", 3, 3, 1);
+ ASSERT_OK_AND_ASSIGN(
+ std::vector<DataEvolutionSplitRead::VectorReadRange> ranges,
+ DataEvolutionSplitRead::PlanVectorReadRanges({main, vector}, {1}, EmbeddingFieldIds));
+ std::vector<std::vector<std::shared_ptr<DataFileMeta>>> expected(10, {main});
+ for (int32_t i : {3, 4, 5}) {
+ expected[i] = {main, vector};
+ }
+ CheckVectorReadProviders(ranges, 0, expected);
+}
+
+TEST_F(DataEvolutionSplitReadTest, TestPlanVectorReadRangesForScalarProjection) {
+ auto main = CreateNormalFile("main.parquet", 0, 10, 0);
+ auto vector0 = CreateDataFileMeta("vector0.vector.vortex", 0, 3, 1);
+ auto vector1 = CreateDataFileMeta("vector1.vector.vortex", 3, 7, 1);
+ ASSERT_OK_AND_ASSIGN(std::vector<DataEvolutionSplitRead::VectorReadRange> ranges,
+ DataEvolutionSplitRead::PlanVectorReadRanges({vector1, main, vector0}, {0},
+ EmbeddingFieldIds));
+ std::vector<std::vector<std::shared_ptr<DataFileMeta>>> expected(10, {main});
+ CheckVectorReadProviders(ranges, 0, expected);
+}
+
+TEST_F(DataEvolutionSplitReadTest, TestPlanVectorOnlyReadRangesKeepsRowCountProviders) {
+ auto other_field = CreateDataFileMeta("other.vector.vortex", 0, 10, 1);
+ auto projected_field = CreateDataFileMeta("projected.vector.vortex", 3, 3, 2);
+ auto file_to_field_ids =
+ [&other_field](const std::shared_ptr<DataFileMeta>& file) -> Result<std::vector<int32_t>> {
+ return file == other_field ? std::vector<int32_t>{2} : std::vector<int32_t>{1};
+ };
+ ASSERT_OK_AND_ASSIGN(std::vector<DataEvolutionSplitRead::VectorReadRange> ranges,
+ DataEvolutionSplitRead::PlanVectorReadRanges(
+ {other_field, projected_field}, {1}, file_to_field_ids));
+ std::vector<std::vector<std::shared_ptr<DataFileMeta>>> expected(10, {other_field});
+ for (int32_t i : {3, 4, 5}) {
+ expected[i] = {projected_field};
+ }
+ CheckVectorReadProviders(ranges, 0, expected);
+}
+
+TEST_F(DataEvolutionSplitReadTest, TestPlanVectorReadRangesUsesFieldIdsAfterRename) {
+ auto main = CreateNormalFile("main.parquet", 0, 10, 0);
+ auto old_vector = CreateDataFileMeta("old.vector.vortex", 0, 10, 1);
+ old_vector->write_cols = std::vector<std::string>{"embedding_before_rename"};
+ old_vector->schema_id = 0;
+ auto new_vector = CreateDataFileMeta("new.vector.vortex", 3, 3, 2);
+ new_vector->write_cols = std::vector<std::string>{"embedding_after_rename"};
+ new_vector->schema_id = 1;
+ auto file_to_field_ids =
+ [](const std::shared_ptr<DataFileMeta>& file) -> Result<std::vector<int32_t>> {
+ const std::map<int64_t, std::string> schema_names = {{0, "embedding_before_rename"},
+ {1, "embedding_after_rename"}};
+ if (file->write_cols != std::vector<std::string>{schema_names.at(file->schema_id)}) {
+ return Status::Invalid("VECTOR write column does not match its original schema");
+ }
+ return std::vector<int32_t>{1};
+ };
+ ASSERT_OK_AND_ASSIGN(std::vector<DataEvolutionSplitRead::VectorReadRange> ranges,
+ DataEvolutionSplitRead::PlanVectorReadRanges(
+ {main, old_vector, new_vector}, {1}, file_to_field_ids));
+ std::vector<std::vector<std::shared_ptr<DataFileMeta>>> expected(10, {main, old_vector});
+ for (int32_t i : {3, 4, 5}) {
+ expected[i] = {main, new_vector};
+ }
+ CheckVectorReadProviders(ranges, 0, expected);
+}
+
TEST_F(DataEvolutionSplitReadTest, TestSplitWithSameFirstRowId) {
std::vector<std::shared_ptr<DataFileMeta>> files = {CreateDataFileMeta("file0", 1l, 1, 10),
CreateDataFileMeta("file1", 1l, 1, 20),
diff --git a/src/paimon/core/schema/schema_validation.cpp b/src/paimon/core/schema/schema_validation.cpp
index 694a4e5c..3a73838d 100644
--- a/src/paimon/core/schema/schema_validation.cpp
+++ b/src/paimon/core/schema/schema_validation.cpp
@@ -120,10 +120,11 @@ Status ValidateSharedShreddingFileFormat(const std::string& option_key,
Status ValidateVectorFileFormat(const std::string& option_key, const std::string& file_format) {
if (!StringUtils::EqualsIgnoreCase(file_format, "parquet") &&
- !StringUtils::EqualsIgnoreCase(file_format, "lance")) {
- return Status::Invalid(
- fmt::format("VECTOR currently only supports parquet/lance data files, but {} is {}.",
- option_key, file_format));
+ !StringUtils::EqualsIgnoreCase(file_format, "lance") &&
+ !StringUtils::EqualsIgnoreCase(file_format, "vortex")) {
+ return Status::Invalid(fmt::format(
+ "VECTOR currently only supports parquet/lance/vortex data files, but {} is {}.",
+ option_key, file_format));
}
return Status::OK();
}
@@ -397,6 +398,7 @@ Status SchemaValidation::ValidateTableSchema(const TableSchema& schema) {
PAIMON_RETURN_NOT_OK(ValidateVortexDataFields(schema, options));
PAIMON_RETURN_NOT_OK(ValidateMapStorageLayout(schema, options));
PAIMON_RETURN_NOT_OK(ValidateVectorFields(schema, options));
+ PAIMON_RETURN_NOT_OK(ValidateVectorStore(schema, options));
return Status::OK();
}
@@ -1261,6 +1263,57 @@ Status SchemaValidation::ValidateMapStorageLayout(const TableSchema& schema,
return Status::OK();
}
+Status SchemaValidation::ValidateVectorStore(const TableSchema& schema,
+ const CoreOptions& options) {
+ const std::shared_ptr<FileFormat> vector_format = options.GetVectorFileFormat();
+ if (!vector_format) {
+ return Status::OK();
+ }
+ PAIMON_RETURN_NOT_OK(
+ ValidateVectorFileFormat(Options::VECTOR_FILE_FORMAT, vector_format->Identifier()));
+ if (!schema.PrimaryKeys().empty()) {
+ return Status::NotImplemented("Dedicated VECTOR storage only supports append-only tables");
+ }
+ if (!options.RowTrackingEnabled() || !options.DataEvolutionEnabled()) {
+ return Status::Invalid(
+ "Dedicated VECTOR storage requires row-tracking.enabled and data-evolution.enabled");
+ }
+ if (options.GetBucket() != -1) {
+ return Status::Invalid("Dedicated VECTOR storage requires bucket=-1");
+ }
+ if (options.GetFileFormat()->Identifier() != "parquet" ||
+ options.GetWriteFileFormat(/*level=*/0)->Identifier() != "parquet") {
+ return Status::NotImplemented(
+ "Dedicated VECTOR storage currently requires parquet main files");
+ }
+ int32_t vector_count = 0;
+ int32_t normal_count = 0;
+ for (const DataField& field : schema.Fields()) {
+ if (ContainsBlobField(field.ArrowField())) {
+ return Status::NotImplemented(
+ "Dedicated VECTOR storage does not yet support BLOB columns");
+ }
+ if (VectorUtils::ContainsVectorField(field.ArrowField())) {
+ if (field.Type()->id() != arrow::Type::FIXED_SIZE_LIST) {
+ return Status::NotImplemented(
+ "Dedicated VECTOR storage only supports a top-level VECTOR column");
+ }
+ ++vector_count;
+ } else if (std::find(schema.PartitionKeys().begin(), schema.PartitionKeys().end(),
+ field.Name()) == schema.PartitionKeys().end()) {
+ ++normal_count;
+ }
+ }
+ if (vector_count != 1) {
+ return Status::NotImplemented(
+ "Dedicated VECTOR storage requires exactly one top-level VECTOR column");
+ }
+ if (normal_count == 0) {
+ return Status::Invalid("Dedicated VECTOR storage requires a normal non-partition column");
+ }
+ return Status::OK();
+}
+
Status SchemaValidation::ValidateVectorFields(const TableSchema& schema,
const CoreOptions& options) {
bool has_vector = false;
diff --git a/src/paimon/core/schema/schema_validation.h b/src/paimon/core/schema/schema_validation.h
index 8c1e41f3..813aaebb 100644
--- a/src/paimon/core/schema/schema_validation.h
+++ b/src/paimon/core/schema/schema_validation.h
@@ -55,6 +55,9 @@ class SchemaValidation {
/// table schema options.
static Status ValidateRealtime(const TableSchema& schema, const CoreOptions& options);
+ /// Checks the deliberately limited dedicated-vector write path, including runtime overrides.
+ static Status ValidateVectorStore(const TableSchema& schema, const CoreOptions& options);
+
/// Validates a schema a table is about to be created from, whichever catalog is creating it.
///
/// It picks the rules by the schema's `type` option: a format table owns no bucket, manifest
diff --git a/src/paimon/core/schema/schema_validation_test.cpp b/src/paimon/core/schema/schema_validation_test.cpp
index d25aa0fa..d7c380ff 100644
--- a/src/paimon/core/schema/schema_validation_test.cpp
+++ b/src/paimon/core/schema/schema_validation_test.cpp
@@ -237,7 +237,7 @@ TEST(SchemaValidationTest, TestVectorType) {
TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{},
/*primary_keys=*/{}, orc_options));
ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
- "VECTOR currently only supports parquet/lance data files");
+ "VECTOR currently only supports parquet/lance/vortex data files");
std::map<std::string, std::string> changelog_options = {
{Options::BUCKET, "1"},
@@ -338,7 +338,7 @@ TEST(SchemaValidationTest, TestVectorType) {
TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{},
/*primary_keys=*/{}, data_evolution_options));
ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
- "VECTOR currently only supports parquet/lance data files");
+ "VECTOR currently only supports parquet/lance/vortex data files");
}
#ifdef PAIMON_ENABLE_MOSAIC
@@ -515,7 +515,106 @@ TEST(SchemaValidationTest, TestLanceDataTypes) {
}
#endif
+TEST(SchemaValidationTest, TestDedicatedVectorStore) {
+ auto vector_type =
+ arrow::fixed_size_list(arrow::field("item", arrow::float32(), /*nullable=*/false), 3);
+ auto id = arrow::field("id", arrow::int64());
+ auto embedding = arrow::field("embedding", vector_type);
+ std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+ {Options::FILE_FORMAT, "parquet"},
+ {Options::VECTOR_FILE_FORMAT, "parquet"},
+ {Options::ROW_TRACKING_ENABLED, "true"},
+ {Options::DATA_EVOLUTION_ENABLED, "true"}};
+ auto validate = [&](const arrow::FieldVector& fields,
+ const std::map<std::string, std::string>& opts) -> Status {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<TableSchema> table_schema,
+ TableSchema::Create(0, arrow::schema(fields), {}, {}, opts));
+ return SchemaValidation::ValidateTableSchema(*table_schema);
+ };
+ ASSERT_OK(validate({id, embedding}, options));
+ ASSERT_NOK_WITH_MSG(validate({id}, options), "exactly one top-level VECTOR");
+ ASSERT_NOK_WITH_MSG(validate({embedding}, options), "normal non-partition column");
+ ASSERT_NOK_WITH_MSG(validate({id, embedding, arrow::field("other", vector_type)}, options),
+ "exactly one top-level VECTOR");
+ ASSERT_NOK_WITH_MSG(
+ validate({id, arrow::field("nested", arrow::struct_({embedding}))}, options),
+ "top-level VECTOR");
+ ASSERT_NOK_WITH_MSG(validate({id, embedding, BlobUtils::ToArrowField("blob", false)}, options),
+ "does not yet support BLOB columns");
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<TableSchema> partition_only_schema,
+ TableSchema::Create(0, arrow::schema({id, embedding}), {"id"}, {}, options));
+ ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*partition_only_schema),
+ "normal non-partition column");
+ auto invalid_options = options;
+ invalid_options[Options::ROW_TRACKING_ENABLED] = "false";
+ ASSERT_NOK(validate({id, embedding}, invalid_options));
+ invalid_options = options;
+ invalid_options[Options::BUCKET] = "1";
+ ASSERT_NOK(validate({id, embedding}, invalid_options));
+ invalid_options = options;
+ invalid_options[Options::DATA_EVOLUTION_ENABLED] = "false";
+ ASSERT_NOK_WITH_MSG(validate({id, embedding}, invalid_options),
+ "requires row-tracking.enabled and data-evolution.enabled");
+#ifdef PAIMON_ENABLE_VORTEX
+ invalid_options = options;
+ invalid_options[Options::FILE_FORMAT] = "vortex";
+ ASSERT_NOK_WITH_MSG(validate({id, embedding}, invalid_options), "requires parquet main files");
+ invalid_options = options;
+ invalid_options[Options::FILE_FORMAT_PER_LEVEL] = "0:vortex";
+ ASSERT_NOK_WITH_MSG(validate({id, embedding}, invalid_options), "requires parquet main files");
+#endif
+ invalid_options = options;
+ invalid_options[Options::VECTOR_FILE_FORMAT] = "avro";
+ ASSERT_NOK_WITH_MSG(validate({id, embedding}, invalid_options),
+ "VECTOR currently only supports parquet/lance/vortex");
+ invalid_options = options;
+ invalid_options[Options::VECTOR_TARGET_FILE_SIZE] = "0 b";
+ ASSERT_NOK_WITH_MSG(validate({id, embedding}, invalid_options), "must be positive");
+
+ ASSERT_OK_AND_ASSIGN(CoreOptions core_options, CoreOptions::FromMap(options));
+ ASSERT_EQ(core_options.GetVectorTargetFileSize(),
+ core_options.GetTargetFileSize(/*has_primary_key=*/false));
+ options[Options::VECTOR_TARGET_FILE_SIZE] = "1 mb";
+ ASSERT_OK_AND_ASSIGN(core_options, CoreOptions::FromMap(options));
+ ASSERT_EQ(core_options.GetVectorTargetFileSize(), 1024 * 1024);
+
+ std::vector<std::string> vector_formats = {"parquet"};
+#ifdef PAIMON_ENABLE_LANCE
+ vector_formats.push_back("lance");
+#endif
+#ifdef PAIMON_ENABLE_VORTEX
+ vector_formats.push_back("vortex");
+#endif
+ for (const auto& format : vector_formats) {
+ SCOPED_TRACE(format);
+ options[Options::VECTOR_FILE_FORMAT] = format;
+ ASSERT_OK(validate({id, embedding}, options));
+ }
+}
+
#ifdef PAIMON_ENABLE_VORTEX
+TEST(SchemaValidationTest, TestVortexVectorType) {
+ auto vector_type =
+ arrow::fixed_size_list(arrow::field("item", arrow::float32(), /*nullable=*/false), 3);
+ auto schema =
+ arrow::schema({arrow::field("id", arrow::int64()), arrow::field("embedding", vector_type)});
+ for (const auto& [option_key, option_value] : std::vector<std::pair<std::string, std::string>>{
+ {Options::FILE_FORMAT, "vortex"},
+ {Options::FILE_FORMAT, "VORTEX"},
+ {Options::FILE_FORMAT_PER_LEVEL, "1:vortex"},
+ {Options::CHANGELOG_FILE_FORMAT, "vortex"}}) {
+ SCOPED_TRACE(option_key + "=" + option_value);
+ std::map<std::string, std::string> options = {{Options::BUCKET, "1"},
+ {Options::FILE_FORMAT, "parquet"}};
+ options[option_key] = option_value;
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+ TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{},
+ /*primary_keys=*/{"id"}, options));
+ ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+ }
+}
+
TEST(SchemaValidationTest, TestVortexDataTypes) {
std::shared_ptr<TableSchema> table_schema;
// Vortex as the default file format: unsupported types are rejected at validation time.
diff --git a/src/paimon/core/table/table.cpp b/src/paimon/core/table/table.cpp
index bc708552..944ffab2 100644
--- a/src/paimon/core/table/table.cpp
+++ b/src/paimon/core/table/table.cpp
@@ -317,6 +317,11 @@ Result<std::unique_ptr<FileStoreWrite>> FileStoreWrite::Create(std::unique_ptr<W
CoreOptions options,
CoreOptions::FromMap(opts, specific_fs, ctx->GetFileSystemSchemeToIdentifierMap()));
PAIMON_RETURN_NOT_OK(SchemaValidation::ValidateRealtime(*schema, options));
+ PAIMON_RETURN_NOT_OK(SchemaValidation::ValidateVectorStore(*schema, options));
+ if (options.GetVectorFileFormat() && !ctx->GetWriteSchema().empty()) {
+ return Status::NotImplemented(
+ "Dedicated VECTOR storage does not yet support custom write schemas or partial writes");
+ }
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Schema> partition_schema,
FieldMapping::GetPartitionSchema(arrow_schema, schema->PartitionKeys()));
diff --git a/src/paimon/format/vortex/vortex_file_format_test.cpp b/src/paimon/format/vortex/vortex_file_format_test.cpp
index 59bb7189..cb383f74 100644
--- a/src/paimon/format/vortex/vortex_file_format_test.cpp
+++ b/src/paimon/format/vortex/vortex_file_format_test.cpp
@@ -450,6 +450,99 @@ TEST_F(VortexFileFormatTest, WriteThenReadFixedSizeListOfString) {
AssertReadWithBatchSizes(path, schema, expected, {1, 2, 3, 4});
}
+TEST_F(VortexFileFormatTest, WriteThenReadVectors) {
+ const std::vector<std::shared_ptr<arrow::DataType>> element_types = {
+ arrow::boolean(), arrow::int8(), arrow::int16(), arrow::int32(),
+ arrow::int64(), arrow::float32(), arrow::float64()};
+ for (const auto& element_type : element_types) {
+ SCOPED_TRACE(element_type->ToString());
+ std::string path =
+ PathUtil::JoinPath(directory_->Str(), "vector-" + element_type->ToString() + ".vortex");
+ auto vector_type =
+ arrow::fixed_size_list(arrow::field("item", element_type, /*nullable=*/false), 3);
+ arrow::FieldVector fields = {arrow::field("embedding", vector_type)};
+ auto schema = arrow::schema(fields);
+ std::string json = R"([
+ [[1, 2, 3]], [null], [[4, 5, 6]], [[-1, 0, 1]], [[7, 8, 9]]
+ ])";
+ if (element_type->id() == arrow::Type::BOOL) {
+ json = R"([
+ [[true, false, true]], [null], [[false, true, false]],
+ [[true, true, false]], [[false, false, true]]
+ ])";
+ } else if (element_type->id() == arrow::Type::FLOAT ||
+ element_type->id() == arrow::Type::DOUBLE) {
+ json = R"([
+ [[1.25, 2.5, 3.75]], [null], [[4.5, 5.25, 6.75]],
+ [[-1.5, 0, 1.5]], [[7.25, 8.5, 9.75]]
+ ])";
+ }
+ std::shared_ptr<arrow::Array> expected =
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), json).ValueOrDie();
+
+ ASSERT_OK(WriteFile(path, schema, expected, /*batch_size=*/2));
+ AssertReadWithBatchSizes(path, schema, expected, {1, 2, 3, 8});
+ }
+}
+
+TEST_F(VortexFileFormatTest, RejectNullVectorElements) {
+ auto vector_type =
+ arrow::fixed_size_list(arrow::field("item", arrow::float32(), /*nullable=*/false), 3);
+ auto schema = arrow::schema({arrow::field("embedding", vector_type)});
+ auto array = arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(schema->fields()),
+ R"([[[1, null, 3]]])")
+ .ValueOrDie();
+ ASSERT_NOK_WITH_MSG(WriteFile(PathUtil::JoinPath(directory_->Str(), "invalid-vector.vortex"),
+ schema, array, /*batch_size=*/1),
+ "VECTOR cannot contain null elements");
+}
+
+TEST_F(VortexFileFormatTest, WriteThenReadVectorsWithNullAncestors) {
+ auto vector_type =
+ arrow::fixed_size_list(arrow::field("item", arrow::float32(), /*nullable=*/false), 3);
+ auto vectors = arrow::ipc::internal::json::ArrayFromJSON(
+ vector_type, R"([[1, null, 3], [4, 5, 6], [7, 8, 9], [10, null, 12]])")
+ .ValueOrDie();
+ auto validity = arrow::Buffer::FromString(std::string(1, '\x06'));
+ for (bool list_parent : {false, true}) {
+ SCOPED_TRACE(list_parent);
+ std::shared_ptr<arrow::Array> parent;
+ if (list_parent) {
+ auto offsets =
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::int32(), "[0, 1, 2, 3, 4]")
+ .ValueOrDie();
+ parent = std::make_shared<arrow::ListArray>(arrow::list(vector_type), 4,
+ offsets->data()->buffers[1], vectors,
+ validity, /*null_count=*/2);
+ } else {
+ parent = arrow::StructArray::Make({vectors}, {arrow::field("embedding", vector_type)},
+ validity, /*null_count=*/2)
+ .ValueOrDie();
+ }
+ auto schema = arrow::schema({arrow::field("payload", parent->type())});
+ auto input = arrow::StructArray::Make({parent}, schema->fields()).ValueOrDie();
+ ASSERT_TRUE(input->ValidateFull().ok());
+ auto expected = arrow::ipc::internal::json::ArrayFromJSON(
+ arrow::struct_(schema->fields()),
+ R"([[null], [[[4, 5, 6]]], [[[7, 8, 9]]], [null]])")
+ .ValueOrDie();
+ std::string path =
+ PathUtil::JoinPath(directory_->Str(), list_parent ? "null-list-vector.vortex"
+ : "null-struct-vector.vortex");
+ ASSERT_OK(WriteFile(path, schema, input, /*batch_size=*/2));
+ AssertReadWithBatchSizes(path, schema, expected, {1, 2, 3, 5});
+
+ // The same elements are invalid when their ancestors are visible.
+ auto visible_parent = parent->data()->Copy();
+ visible_parent->buffers[0].reset();
+ visible_parent->SetNullCount(0);
+ input = arrow::StructArray::Make({arrow::MakeArray(visible_parent)}, schema->fields())
+ .ValueOrDie();
+ ASSERT_NOK_WITH_MSG(WriteFile(path + ".invalid", schema, input, /*batch_size=*/2),
+ "VECTOR cannot contain null elements");
+ }
+}
+
// Regression for a struct nested inside a list: the LIST branch recurses into the struct, whose
// utf8 leaf is a view; the combination must normalize back to list<struct<utf8, int32>>.
TEST_F(VortexFileFormatTest, WriteThenReadListOfStruct) {
@@ -723,6 +816,46 @@ TEST_F(VortexFileFormatTest, WriteThenReadSlicedStructColumn) {
AssertReadWithBatchSizes(path, schema, expected, {1, 2, 4});
}
+TEST_F(VortexFileFormatTest, RollingSizeIncludesBufferedBatches) {
+ const std::string path = PathUtil::JoinPath(directory_->Str(), "rolling-size.vortex");
+ auto vector_type =
+ arrow::fixed_size_list(arrow::field("item", arrow::float32(), /*nullable=*/false), 3);
+ auto schema = arrow::schema({arrow::field("id", arrow::int32(), /*nullable=*/false),
+ arrow::field("embedding", vector_type, /*nullable=*/false)});
+ std::shared_ptr<arrow::Array> written =
+ arrow::ipc::internal::json::ArrayFromJSON(
+ arrow::struct_(schema->fields()),
+ R"([[1,[1,2,3]],[2,[2,3,4]],[3,[3,4,5]],[4,[4,5,6]]])")
+ .ValueOrDie();
+ ::ArrowSchema ffi_schema = {};
+ ASSERT_TRUE(arrow::ExportSchema(*schema, &ffi_schema).ok());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<WriterBuilder> writer_builder,
+ format_->CreateWriterBuilder(&ffi_schema, /*batch_size=*/2));
+ writer_builder->WithMemoryPool(pool_);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<OutputStream> output,
+ file_system_->Create(path, /*overwrite=*/false));
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FormatWriter> writer,
+ writer_builder->Build(output, "zstd"));
+ ASSERT_OK_AND_ASSIGN(bool full, writer->ReachTargetSize(/*suggested_check=*/true, 1));
+ ASSERT_FALSE(full);
+ for (int64_t offset : {0, 2}) {
+ ::ArrowArray ffi_array = {};
+ ASSERT_TRUE(arrow::ExportArray(*written->Slice(offset, 2), &ffi_array).ok());
+ ASSERT_OK(writer->AddBatch(&ffi_array));
+ ASSERT_OK_AND_ASSIGN(full, writer->ReachTargetSize(/*suggested_check=*/false, 1));
+ ASSERT_FALSE(full);
+ // Each two-row slice contributes 32 input bytes; Java estimates a compression ratio of 1/4.
+ const int64_t target = (offset + 2) * 4;
+ ASSERT_OK_AND_ASSIGN(full, writer->ReachTargetSize(/*suggested_check=*/true, target));
+ ASSERT_TRUE(full);
+ ASSERT_OK_AND_ASSIGN(full, writer->ReachTargetSize(/*suggested_check=*/true, target + 1));
+ ASSERT_FALSE(full);
+ }
+ ASSERT_OK(writer->Finish());
+ ASSERT_OK(output->Close());
+ AssertReadWithBatchSizes(path, schema, written, {1, 3});
+}
+
// Regression: Flush() between batches must be safe (it is serialized with the background writer
// task's writes on the host OutputStream) and must not lose or reorder any data.
TEST_F(VortexFileFormatTest, FlushBetweenBatchesThenRead) {
diff --git a/src/paimon/format/vortex/vortex_format_writer.cpp b/src/paimon/format/vortex/vortex_format_writer.cpp
index 47d27111..ffa9be25 100644
--- a/src/paimon/format/vortex/vortex_format_writer.cpp
+++ b/src/paimon/format/vortex/vortex_format_writer.cpp
@@ -21,18 +21,93 @@
#include <utility>
+#include "arrow/array/array_nested.h"
+#include "arrow/array/data.h"
#include "arrow/c/bridge.h"
+#include "arrow/compute/api.h"
#include "arrow/memory_pool.h"
+#include "arrow/scalar.h"
#include "arrow/type.h"
+#include "arrow/util/byte_size.h"
#include "paimon/common/metrics/metrics_impl.h"
#include "paimon/common/utils/arrow/arrow_utils.h"
#include "paimon/common/utils/arrow/mem_utils.h"
#include "paimon/common/utils/arrow/status_utils.h"
+#include "paimon/common/utils/arrow/vector_utils.h"
+#include "paimon/common/utils/checked_cast.h"
#include "paimon/core/schema/schema_validation.h"
#include "paimon/format/vortex/vortex_ffi_util.h"
#include "paimon/fs/file_system.h"
namespace paimon::vortex {
+namespace {
+
+constexpr double COMPRESSION_RATIO = 0.25;
+
+Status ValidateVisibleVectorElements(const std::shared_ptr<arrow::Array>& array,
+ arrow::MemoryPool* pool) {
+ if (!VectorUtils::ContainsVectorType(array->type())) {
+ return Status::OK();
+ }
+ switch (array->type_id()) {
+ case arrow::Type::STRUCT: {
+ const auto& rows = checked_cast<const arrow::StructArray&>(*array);
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(arrow::ArrayVector fields, rows.Flatten(pool));
+ for (const auto& field : fields) {
+ PAIMON_RETURN_NOT_OK(ValidateVisibleVectorElements(field, pool));
+ }
+ return Status::OK();
+ }
+ case arrow::Type::LIST: {
+ const auto& list = checked_cast<const arrow::ListArray&>(*array);
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Array> values,
+ list.Flatten(pool));
+ return ValidateVisibleVectorElements(values, pool);
+ }
+ case arrow::Type::FIXED_SIZE_LIST: {
+ const auto& list = checked_cast<const arrow::FixedSizeListArray&>(*array);
+ if (!array->type()->field(0)->nullable()) {
+ PAIMON_RETURN_NOT_OK(VectorUtils::ValidateVectorElements(list));
+ }
+ if (!VectorUtils::ContainsVectorType(list.value_type())) {
+ return Status::OK();
+ }
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Array> values,
+ list.Flatten(pool));
+ return ValidateVisibleVectorElements(values, pool);
+ }
+ default:
+ return Status::OK();
+ }
+}
+
+Result<std::shared_ptr<arrow::ArrayData>> NormalizeVectorNullElements(
+ const std::shared_ptr<arrow::ArrayData>& data, arrow::compute::ExecContext* context) {
+ std::shared_ptr<arrow::ArrayData> normalized = data;
+ for (size_t i = 0; i < data->child_data.size(); ++i) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::ArrayData> child,
+ NormalizeVectorNullElements(data->child_data[i], context));
+ if (data->type->id() == arrow::Type::FIXED_SIZE_LIST && !data->type->field(0)->nullable() &&
+ child->GetNullCount() != 0) {
+ // Null VECTORs or ancestors may own null slots. Vortex imports these masked slots too.
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Scalar> zero,
+ arrow::MakeScalar(child->type, 0));
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ arrow::Datum filled,
+ arrow::compute::CallFunction("coalesce", {arrow::MakeArray(child), zero}, context));
+ child = filled.make_array()->data();
+ }
+ if (child != data->child_data[i]) {
+ if (normalized == data) {
+ normalized = data->Copy();
+ }
+ normalized->child_data[i] = child;
+ }
+ }
+ return normalized;
+}
+
+} // namespace
VortexFormatWriter::VortexFormatWriter(std::shared_ptr<OutputStream> output,
std::shared_ptr<arrow::Schema> schema, VxSessionPtr session,
@@ -103,6 +178,13 @@ Status VortexFormatWriter::AddBatch(::ArrowArray* batch) {
arrow::ImportArray(batch, arrow::struct_(schema_->fields())));
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> normalized,
ArrowUtils::NormalizeArrayOffsets(imported, arrow_pool_.get()));
+ PAIMON_RETURN_NOT_OK(ValidateVisibleVectorElements(normalized, arrow_pool_.get()));
+ arrow::compute::ExecContext context(arrow_pool_.get());
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::ArrayData> normalized_data,
+ NormalizeVectorNullElements(normalized->data(), &context));
+ normalized = arrow::MakeArray(normalized_data);
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(int64_t batch_bytes,
+ arrow::util::ReferencedBufferSize(*normalized));
PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*normalized, &rebased_batch));
// The rebased buffers belong to arrow_pool_, which dies with this writer; keep it alive until
// Rust releases the exported array.
@@ -122,6 +204,7 @@ Status VortexFormatWriter::AddBatch(::ArrowArray* batch) {
return VortexCallbackError("push batch to Vortex sink", error,
output_context_->GetCallbackStatus());
}
+ ffi_bytes_ += batch_bytes;
return Status::OK();
}
@@ -156,7 +239,8 @@ Result<bool> VortexFormatWriter::ReachTargetSize(bool suggested_check, int64_t t
if (!suggested_check || output_context_ == nullptr) {
return false;
}
- return output_context_->BytesWritten() >= target_size;
+ // Match Java's estimate: background writes may still be buffered when rolling is checked.
+ return static_cast<int64_t>(ffi_bytes_ * COMPRESSION_RATIO) >= target_size;
}
std::shared_ptr<Metrics> VortexFormatWriter::GetWriterMetrics() const {
diff --git a/src/paimon/format/vortex/vortex_format_writer.h b/src/paimon/format/vortex/vortex_format_writer.h
index 3aff6c06..55c38bec 100644
--- a/src/paimon/format/vortex/vortex_format_writer.h
+++ b/src/paimon/format/vortex/vortex_format_writer.h
@@ -72,6 +72,7 @@ class VortexFormatWriter : public FormatWriter {
// RAII alias cannot express. Nulled right after either, so it is never used or freed twice.
vx_callback_sink* sink_;
std::shared_ptr<Metrics> metrics_;
+ int64_t ffi_bytes_ = 0;
bool finished_ = false;
};
diff --git a/test/inte/blob_table_inte_test.cpp b/test/inte/blob_table_inte_test.cpp
index 80ea97e5..eabb28a7 100644
--- a/test/inte/blob_table_inte_test.cpp
+++ b/test/inte/blob_table_inte_test.cpp
@@ -57,6 +57,7 @@
#include "paimon/core/snapshot.h"
#include "paimon/core/stats/simple_stats.h"
#include "paimon/core/table/source/data_split_impl.h"
+#include "paimon/core/table/source/plan_impl.h"
#include "paimon/core/utils/file_utils.h"
#include "paimon/core/utils/snapshot_manager.h"
#include "paimon/data/blob.h"
@@ -1894,12 +1895,32 @@ TEST_P(BlobTableInteTest, TestDataEvolutionBlobPartialUpdateMultipleLayers) {
const std::vector<std::string> expected_rows = {
R"([[1, "a", "update1_0"]])", R"([[2, "b", "blob_1"]])", R"([[3, "c", "blob_2"]])",
R"([[4, "d", "update2_3"]])"};
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> unpruned_plan, ScanTable(table_path));
+ ASSERT_EQ(unpruned_plan->Splits().size(), 1);
+ auto unpruned_split = std::dynamic_pointer_cast<DataSplitImpl>(unpruned_plan->Splits()[0]);
+ ASSERT_NE(unpruned_split, nullptr);
+ ASSERT_EQ(unpruned_split->DataFiles().size(), 4);
for (int32_t i = 0; i < static_cast<int32_t>(expected_rows.size()); i++) {
+ SCOPED_TRACE(i);
auto expected_single = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), expected_rows[i])
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_single,
/*predicate=*/nullptr, /*row_ranges=*/{Range(i, i)}));
+
+ // Range planning can narrow a read after scanning, leaving unselected BLOB files
+ // in the split. Their sequence layers must still align with the selected rows.
+ auto indexed_split =
+ std::make_shared<IndexedSplitImpl>(unpruned_split, std::vector<Range>{Range(i, i)});
+ auto indexed_plan = std::make_shared<PlanImpl>(
+ unpruned_plan->SnapshotId(), std::vector<std::shared_ptr<Split>>{indexed_split});
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> result,
+ ReadTable(table_path, schema->field_names(), indexed_plan));
+ ASSERT_NE(result, nullptr);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::StructArray> expected_with_row_kind,
+ PrependRowKindColumn(expected_single));
+ ASSERT_TRUE(result->Equals(arrow::ChunkedArray(expected_with_row_kind)))
+ << result->ToString();
}
}
diff --git a/test/inte/data_evolution_table_test.cpp b/test/inte/data_evolution_table_test.cpp
index 804e7c26..c508fd19 100644
--- a/test/inte/data_evolution_table_test.cpp
+++ b/test/inte/data_evolution_table_test.cpp
@@ -17,6 +17,7 @@
* under the License.
*/
#include <algorithm>
+#include <iterator>
#include <tuple>
#include "arrow/type.h"
@@ -30,6 +31,7 @@
#include "paimon/common/utils/date_time_utils.h"
#include "paimon/common/utils/path_util.h"
#include "paimon/common/utils/scope_guard.h"
+#include "paimon/common/utils/vector_store_utils.h"
#include "paimon/core/global_index/indexed_split_impl.h"
#include "paimon/core/table/source/data_split_impl.h"
#include "paimon/defs.h"
@@ -1796,6 +1798,442 @@ TEST_P(DataEvolutionTableTest, TestPartitionWithPredicate) {
}
}
+#ifdef PAIMON_ENABLE_VORTEX
+TEST_P(DataEvolutionTableTest, TestReadJavaDedicatedVectorStore) {
+ if (FileFormat() != "parquet") {
+ GTEST_SKIP() << "The Java fixture uses parquet main files";
+ }
+ const std::string fixture_path = PathUtil::JoinPath(
+ GetDataDir(), "vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat");
+ const std::string table_path = PathUtil::JoinPath(dir_->Str(), "java_dedicated_vector");
+ // Preserve the Java-produced files: this test only scans and reads the copied table.
+ ASSERT_TRUE(TestUtil::CopyDirectory(fixture_path, table_path));
+
+ // Java DataTypes.FLOAT() serializes the VECTOR element as nullable in schema-0.
+ auto vector_type =
+ arrow::fixed_size_list(arrow::field("item", arrow::float32(), /*nullable=*/true), 3);
+ arrow::FieldVector fields = {arrow::field("id", arrow::int32()),
+ arrow::field("embedding", vector_type),
+ arrow::field("tag", arrow::utf8())};
+ auto expected = checked_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
+ [1, [1, 2, 3], "a"], [2, null, "b"], [3, [-1, 0, 1], "c"],
+ [4, null, "d"], [5, [7, 8, 9], "e"], [6, [10, 11, 12], "f"]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK(ScanAndRead(table_path, {"id", "embedding", "tag"}, expected));
+ auto normal_projection =
+ arrow::StructArray::Make({expected->field(0), expected->field(2)}, {fields[0], fields[2]})
+ .ValueOrDie();
+ ASSERT_OK(ScanAndRead(table_path, {"id", "tag"}, normal_projection));
+ auto vector_projection =
+ arrow::StructArray::Make({expected->field(1)}, {fields[1]}).ValueOrDie();
+ ASSERT_OK(ScanAndRead(table_path, {"embedding"}, vector_projection));
+
+ // Row ids 1-4 cross the two Java commits and include a null vector from each commit.
+ auto selected_rows = checked_pointer_cast<arrow::StructArray>(expected->Slice(1, 4));
+ ASSERT_OK(ScanAndRead(table_path, {"id", "embedding", "tag"}, selected_rows,
+ /*predicate=*/nullptr, /*row_ranges=*/{Range(1, 4)}));
+ auto selected_vectors =
+ checked_pointer_cast<arrow::StructArray>(vector_projection->Slice(1, 4));
+ ASSERT_OK(ScanAndRead(table_path, {"embedding"}, selected_vectors,
+ /*predicate=*/nullptr, /*row_ranges=*/{Range(1, 4)}));
+
+ ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<Split>> splits, PlanSplits(table_path));
+ std::map<int64_t, std::vector<std::shared_ptr<DataFileMeta>>> row_groups;
+ for (const auto& split : splits) {
+ auto data_split = std::dynamic_pointer_cast<DataSplitImpl>(split);
+ ASSERT_NE(data_split, nullptr);
+ for (const auto& file : data_split->DataFiles()) {
+ ASSERT_TRUE(file->first_row_id.has_value());
+ row_groups[file->first_row_id.value()].push_back(file);
+ }
+ }
+ ASSERT_EQ(row_groups.size(), 2);
+ ASSERT_EQ(row_groups.count(0), 1);
+ ASSERT_EQ(row_groups.count(3), 1);
+ for (const auto& [first_row_id, files] : row_groups) {
+ SCOPED_TRACE(first_row_id);
+ ASSERT_EQ(files.size(), 2);
+ int32_t vector_file_count = 0;
+ for (const auto& file : files) {
+ ASSERT_EQ(file->row_count, 3);
+ if (VectorStoreUtils::IsVectorStoreFile(file->file_name)) {
+ ++vector_file_count;
+ ASSERT_TRUE(StringUtils::EndsWith(file->file_name, ".vector.vortex"));
+ ASSERT_EQ(file->write_cols, std::vector<std::string>({"embedding"}));
+ } else {
+ ASSERT_TRUE(StringUtils::EndsWith(file->file_name, ".parquet"));
+ ASSERT_EQ(file->write_cols, std::vector<std::string>({"id", "tag"}));
+ }
+ }
+ ASSERT_EQ(vector_file_count, 1);
+ }
+}
+
+TEST_P(DataEvolutionTableTest, TestReadJavaIndependentlyRolledVectorStore) {
+ if (FileFormat() != "parquet") {
+ GTEST_SKIP() << "The Java fixture uses parquet main files";
+ }
+ const std::string fixture_path = PathUtil::JoinPath(
+ GetDataDir(), "vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling");
+ const std::string table_path = PathUtil::JoinPath(dir_->Str(), "java_rolled_vector");
+ ASSERT_TRUE(TestUtil::CopyDirectory(fixture_path, table_path));
+
+ constexpr int32_t kRowCount = 3001;
+ auto vector_type = arrow::fixed_size_list(arrow::float32(), 3);
+ arrow::FieldVector fields = {arrow::field("id", arrow::int32()),
+ arrow::field("embedding", vector_type),
+ arrow::field("tag", arrow::utf8())};
+ const std::vector<int32_t> null_vector_ids = {0, 999, 1000, 1999, 2000, 2999, 3000};
+ auto expected = PrepareBulkData(
+ kRowCount,
+ [&null_vector_ids](int32_t id) {
+ std::string vector_json = std::find(null_vector_ids.begin(), null_vector_ids.end(),
+ id) != null_vector_ids.end()
+ ? "null"
+ : fmt::format("[{}, {}, {}]", id, id + 1, id + 2);
+ return fmt::format(R"({}, {}, "row-{}")", id, vector_json, id);
+ },
+ fields);
+ // The Java fixture sets read.batch-size=127 so batch and vector file boundaries differ.
+ auto check_projections = [&](const std::shared_ptr<arrow::StructArray>& rows,
+ const std::vector<Range>& row_ranges) -> Status {
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ std::shared_ptr<arrow::StructArray> normal_projection,
+ arrow::StructArray::Make({rows->field(0), rows->field(2)}, {fields[0], fields[2]}));
+ PAIMON_RETURN_NOT_OK(ScanAndRead(table_path, {"id", "tag"}, normal_projection,
+ /*predicate=*/nullptr, row_ranges));
+ PAIMON_RETURN_NOT_OK(ScanAndRead(table_path, {"id", "embedding", "tag"}, rows,
+ /*predicate=*/nullptr, row_ranges));
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::StructArray> vector_projection,
+ arrow::StructArray::Make({rows->field(1)}, {fields[1]}));
+ return ScanAndRead(table_path, {"embedding"}, vector_projection,
+ /*predicate=*/nullptr, row_ranges);
+ };
+ ASSERT_OK(check_projections(expected, {}));
+
+ ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<Split>> splits, PlanSplits(table_path));
+ std::shared_ptr<DataFileMeta> main_file;
+ std::map<int64_t, std::shared_ptr<DataFileMeta>> vector_files;
+ for (const auto& split : splits) {
+ auto data_split = std::dynamic_pointer_cast<DataSplitImpl>(split);
+ ASSERT_NE(data_split, nullptr);
+ for (const auto& file : data_split->DataFiles()) {
+ ASSERT_TRUE(file->first_row_id.has_value());
+ if (VectorStoreUtils::IsVectorStoreFile(file->file_name)) {
+ ASSERT_TRUE(StringUtils::EndsWith(file->file_name, ".vector.vortex"));
+ ASSERT_EQ(file->write_cols, std::vector<std::string>({"embedding"}));
+ ASSERT_TRUE(vector_files.emplace(file->first_row_id.value(), file).second);
+ } else {
+ ASSERT_EQ(main_file, nullptr);
+ ASSERT_TRUE(StringUtils::EndsWith(file->file_name, ".parquet"));
+ ASSERT_EQ(file->write_cols, std::vector<std::string>({"id", "tag"}));
+ main_file = file;
+ }
+ }
+ }
+ ASSERT_NE(main_file, nullptr);
+ ASSERT_EQ(main_file->first_row_id, 0);
+ ASSERT_EQ(main_file->row_count, kRowCount);
+ ASSERT_GT(vector_files.size(), 1);
+
+ int64_t next_row_id = 0;
+ std::vector<Range> selected_ranges = {Range(0, 0)};
+ for (const auto& [first_row_id, file] : vector_files) {
+ ASSERT_EQ(first_row_id, next_row_id);
+ ASSERT_GT(file->row_count, 0);
+ next_row_id += file->row_count;
+ if (first_row_id == 0) {
+ continue;
+ }
+ SCOPED_TRACE(first_row_id);
+ const int64_t from = std::max<int64_t>(0, first_row_id - 2);
+ const int64_t to = std::min<int64_t>(kRowCount - 1, first_row_id + 2);
+ auto boundary_rows =
+ checked_pointer_cast<arrow::StructArray>(expected->Slice(from, to - from + 1));
+ ASSERT_OK(check_projections(boundary_rows, {Range(from, to)}));
+ selected_ranges.emplace_back(std::max<int64_t>(0, first_row_id - 1),
+ std::min<int64_t>(kRowCount - 1, first_row_id + 1));
+ }
+ ASSERT_EQ(next_row_id, kRowCount);
+
+ // Disjoint selections retain row order while skipping most rows in each physical file.
+ selected_ranges = Range::SortAndMergeOverlap(selected_ranges, /*adjacent=*/true);
+ ASSERT_GT(selected_ranges.size(), 1);
+ arrow::ArrayVector selected_chunks;
+ for (const auto& range : selected_ranges) {
+ selected_chunks.push_back(expected->Slice(range.from, range.Count()));
+ }
+ auto selected_rows =
+ checked_pointer_cast<arrow::StructArray>(arrow::Concatenate(selected_chunks).ValueOrDie());
+ ASSERT_OK(check_projections(selected_rows, selected_ranges));
+
+ // Omit the outer VECTOR files in an in-memory split to exercise NULL fill for uncovered
+ // rows. The main file has no physical embedding column, and the Java fixture stays intact.
+ ASSERT_EQ(splits.size(), 1);
+ auto original_split = std::dynamic_pointer_cast<DataSplitImpl>(splits[0]);
+ ASSERT_NE(original_split, nullptr);
+ auto middle_file = std::next(vector_files.begin())->second;
+ const int64_t middle_from = middle_file->first_row_id.value();
+ const int64_t middle_to = middle_from + middle_file->row_count - 1;
+ DataSplitImpl::Builder gap_builder(
+ original_split->Partition(), original_split->Bucket(), original_split->BucketPath(),
+ std::vector<std::shared_ptr<DataFileMeta>>{main_file, middle_file});
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<DataSplitImpl> gap_split,
+ gap_builder.WithTotalBuckets(original_split->TotalBuckets())
+ .WithSnapshot(original_split->SnapshotId())
+ .RawConvertible(false)
+ .Build());
+ auto gap_expected = PrepareBulkData(
+ kRowCount,
+ [&](int32_t id) {
+ const bool missing_vector = id < middle_from || id > middle_to;
+ const bool null_vector = std::find(null_vector_ids.begin(), null_vector_ids.end(),
+ id) != null_vector_ids.end();
+ std::string vector_json = missing_vector || null_vector
+ ? "null"
+ : fmt::format("[{}, {}, {}]", id, id + 1, id + 2);
+ return fmt::format("0, {}", vector_json);
+ },
+ {SpecialFields::ValueKind().ArrowField(), fields[1]});
+ arrow::ArrayVector selected_gap_chunks;
+ for (const auto& range : selected_ranges) {
+ selected_gap_chunks.push_back(gap_expected->Slice(range.from, range.Count()));
+ }
+ auto selected_gap_expected = checked_pointer_cast<arrow::StructArray>(
+ arrow::Concatenate(selected_gap_chunks).ValueOrDie());
+ ReadContextBuilder read_builder(table_path);
+ read_builder.SetReadFieldNames({"embedding"});
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_builder.Finish());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableRead> read,
+ TableRead::Create(std::move(read_context)));
+ const std::vector<std::pair<std::shared_ptr<Split>, std::shared_ptr<arrow::StructArray>>>
+ gap_cases = {{gap_split, gap_expected},
+ {std::make_shared<IndexedSplitImpl>(gap_split, selected_ranges),
+ selected_gap_expected}};
+ for (const auto& [split, expected_rows] : gap_cases) {
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<BatchReader> reader, read->CreateReader(split));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> result,
+ ReadResultCollector::CollectResult(std::move(reader)));
+ ASSERT_NE(result, nullptr);
+ ASSERT_TRUE(result->Equals(arrow::ChunkedArray(expected_rows)))
+ << "actual=" << result->ToString() << "\nexpected=" << expected_rows->ToString();
+ }
+}
+
+#endif
+
+class DedicatedVectorTableTest : public DataEvolutionTableTest {};
+
+TEST_P(DedicatedVectorTableTest, TestDedicatedVectorStore) {
+ auto vector_type =
+ arrow::fixed_size_list(arrow::field("item", arrow::float32(), /*nullable=*/false), 3);
+ arrow::FieldVector fields = {arrow::field("id", arrow::int32()),
+ arrow::field("embedding", vector_type),
+ arrow::field("tag", arrow::utf8())};
+ std::map<std::string, std::string> options = {
+ {Options::FILE_FORMAT, "parquet"}, {Options::VECTOR_FILE_FORMAT, FileFormat()},
+ {Options::FILE_SYSTEM, "local"}, {Options::BUCKET, "-1"},
+ {Options::ROW_TRACKING_ENABLED, "true"}, {Options::DATA_EVOLUTION_ENABLED, "true"},
+ {Options::TARGET_FILE_ROW_NUM, "2"}, {Options::READ_BATCH_SIZE, "2"}};
+ CreateTable(fields, /*partition_keys=*/{}, options);
+ std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
+ auto input = checked_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
+ [0, [0, 0, 0], "skip"],
+ [1, [1, 2, 3], "a"], [2, null, "b"],
+ [3, null, "c"], [4, null, "d"],
+ [5, [7, 8, 9], "e"]
+ ])")
+ .ValueOrDie());
+ // Sliced input also exercises a pair containing only null vectors.
+ std::vector<std::shared_ptr<arrow::Array>> batches = {input->Slice(1, 2), input->Slice(3, 2),
+ input->Slice(5, 1)};
+ ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> messages,
+ WriteArrays(table_path, {}, {}, batches));
+ ASSERT_EQ(messages.size(), 1);
+ auto message = std::dynamic_pointer_cast<CommitMessageImpl>(messages[0]);
+ ASSERT_NE(message, nullptr);
+ const auto& files = message->data_increment_.new_files_;
+ ASSERT_EQ(files.size(), 6);
+ for (size_t i = 0; i < files.size(); i += 2) {
+ ASSERT_FALSE(VectorStoreUtils::IsVectorStoreFile(files[i]->file_name));
+ ASSERT_TRUE(StringUtils::EndsWith(files[i + 1]->file_name,
+ fmt::format(".vector.{}", FileFormat())));
+ ASSERT_EQ(files[i]->row_count, files[i + 1]->row_count);
+ ASSERT_EQ(files[i]->write_cols, std::vector<std::string>({"id", "tag"}));
+ ASSERT_EQ(files[i + 1]->write_cols, std::vector<std::string>({"embedding"}));
+ }
+ ASSERT_OK(Commit(table_path, messages));
+ auto expected = checked_pointer_cast<arrow::StructArray>(input->Slice(1));
+ ASSERT_OK(ScanAndRead(table_path, {"id", "embedding", "tag"}, expected));
+ auto normal_projection =
+ arrow::StructArray::Make({expected->field(0), expected->field(2)}, {fields[0], fields[2]})
+ .ValueOrDie();
+ ASSERT_OK(ScanAndRead(table_path, {"id", "tag"}, normal_projection));
+ auto vector_projection = checked_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[1]}), R"([
+ [null], [null], [null], [[7, 8, 9]]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK(ScanAndRead(table_path, {"embedding"}, vector_projection, /*predicate=*/nullptr,
+ /*row_ranges=*/{Range(1, 4)}));
+
+ ASSERT_OK_AND_ASSIGN(messages, WriteArray(table_path, {}, input->Slice(1, 2)));
+ ASSERT_OK(Commit(table_path, messages));
+ auto appended_expected = checked_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
+ [1, [1, 2, 3], "a"], [2, null, "b"],
+ [3, null, "c"], [4, null, "d"], [5, [7, 8, 9], "e"],
+ [1, [1, 2, 3], "a"], [2, null, "b"]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK(ScanAndRead(table_path, {"id", "embedding", "tag"}, appended_expected));
+ ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<Split>> splits, PlanSplits(table_path));
+ std::map<int64_t, std::vector<std::shared_ptr<DataFileMeta>>> row_groups;
+ for (const auto& split : splits) {
+ auto data_split = std::dynamic_pointer_cast<DataSplitImpl>(split);
+ ASSERT_NE(data_split, nullptr);
+ for (const auto& file : data_split->DataFiles()) {
+ ASSERT_TRUE(file->first_row_id.has_value());
+ row_groups[file->first_row_id.value()].push_back(file);
+ }
+ }
+ ASSERT_EQ(row_groups.size(), 4);
+ ASSERT_EQ(row_groups.count(0), 1);
+ ASSERT_EQ(row_groups.count(2), 1);
+ ASSERT_EQ(row_groups.count(4), 1);
+ ASSERT_EQ(row_groups.count(5), 1);
+ for (const auto& [first_row_id, group] : row_groups) {
+ ASSERT_EQ(group.size(), 2);
+ ASSERT_EQ(group[0]->row_count, group[1]->row_count);
+ }
+
+ WriteContextBuilder partial_builder(table_path, "partial");
+ partial_builder.WithWriteSchema({"embedding"});
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<WriteContext> partial_context, partial_builder.Finish());
+ ASSERT_NOK_WITH_MSG(FileStoreWrite::Create(std::move(partial_context)).status(),
+ "partial writes");
+
+ WriteContextBuilder invalid_builder(table_path, "invalid");
+ invalid_builder.AddOption(Options::DATA_EVOLUTION_ENABLED, "false");
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<WriteContext> invalid_context, invalid_builder.Finish());
+ ASSERT_NOK_WITH_MSG(FileStoreWrite::Create(std::move(invalid_context)).status(),
+ "requires row-tracking.enabled and data-evolution.enabled");
+}
+
+TEST_P(DedicatedVectorTableTest, TestIndependentVectorRolling) {
+ auto vector_type =
+ arrow::fixed_size_list(arrow::field("item", arrow::float32(), /*nullable=*/false), 3);
+ arrow::FieldVector fields = {arrow::field("id", arrow::int32()),
+ arrow::field("embedding", vector_type),
+ arrow::field("tag", arrow::utf8())};
+ const std::map<std::string, std::string> options = {
+ {Options::FILE_FORMAT, "parquet"}, {Options::VECTOR_FILE_FORMAT, FileFormat()},
+ {Options::FILE_SYSTEM, "local"}, {Options::BUCKET, "-1"},
+ {Options::ROW_TRACKING_ENABLED, "true"}, {Options::DATA_EVOLUTION_ENABLED, "true"},
+ {Options::WRITE_ONLY, "true"}, {Options::TARGET_FILE_SIZE, "128 mb"},
+ {Options::TARGET_FILE_ROW_NUM, "2500"}, {Options::VECTOR_TARGET_FILE_SIZE, "1 b"},
+ {Options::WRITE_BATCH_SIZE, "100"}, {Options::READ_BATCH_SIZE, "127"}};
+ CreateTable(fields, /*partition_keys=*/{}, options);
+ const std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
+ auto input = PrepareBulkData(
+ 6001,
+ [](int32_t id) {
+ const std::string vector = id % 1000 == 0 || id % 1000 == 999
+ ? "null"
+ : fmt::format("[{}, {}, {}]", id, id + 1, id + 2);
+ return fmt::format(R"({}, {}, "row-{}")", id, vector, id);
+ },
+ fields);
+ std::vector<std::shared_ptr<arrow::Array>> batches;
+ for (int64_t offset = 0; offset < input->length(); offset += 1000) {
+ batches.push_back(input->Slice(offset, std::min<int64_t>(1000, input->length() - offset)));
+ }
+ ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> messages,
+ WriteArrays(table_path, {}, {}, batches));
+ ASSERT_EQ(messages.size(), 1);
+ auto message = std::dynamic_pointer_cast<CommitMessageImpl>(messages[0]);
+ ASSERT_NE(message, nullptr);
+ int32_t main_count = 0;
+ int32_t vector_count = 0;
+ int64_t main_rows = 0;
+ int64_t vector_rows = 0;
+ for (const auto& file : message->data_increment_.new_files_) {
+ if (VectorStoreUtils::IsVectorStoreFile(file->file_name)) {
+ ASSERT_GT(main_count, 0);
+ ASSERT_GT(file->row_count, 0);
+ vector_rows += file->row_count;
+ ASSERT_LE(vector_rows, main_rows);
+ ++vector_count;
+ } else {
+ ASSERT_EQ(main_rows, vector_rows);
+ main_rows += file->row_count;
+ ++main_count;
+ }
+ }
+ ASSERT_EQ(main_count, 3);
+ ASSERT_EQ(main_rows, input->length());
+ ASSERT_EQ(vector_rows, main_rows);
+ // Lance checks encoded bytes only; a small file may stay buffered until the main file closes.
+ if (FileFormat() != "lance") {
+ ASSERT_GT(vector_count, main_count);
+ }
+ ASSERT_OK(Commit(table_path, messages));
+
+ ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<Split>> splits, PlanSplits(table_path));
+ std::map<int64_t, std::shared_ptr<DataFileMeta>> main_files;
+ std::map<int64_t, std::shared_ptr<DataFileMeta>> vector_files;
+ for (const auto& split : splits) {
+ auto data_split = std::dynamic_pointer_cast<DataSplitImpl>(split);
+ ASSERT_NE(data_split, nullptr);
+ for (const auto& file : data_split->DataFiles()) {
+ ASSERT_OK_AND_ASSIGN(int64_t first_row_id, file->NonNullFirstRowId());
+ ASSERT_EQ(file->min_sequence_number, 1);
+ ASSERT_EQ(file->max_sequence_number, 1);
+ auto& files =
+ VectorStoreUtils::IsVectorStoreFile(file->file_name) ? vector_files : main_files;
+ ASSERT_TRUE(files.emplace(first_row_id, file).second);
+ }
+ }
+ ASSERT_EQ(main_files.size(), 3);
+ auto vector_file = vector_files.begin();
+ int64_t next_row_id = 0;
+ for (const auto& [first, main] : main_files) {
+ ASSERT_EQ(first, next_row_id);
+ const int64_t end = first + main->row_count;
+ while (next_row_id < end) {
+ ASSERT_NE(vector_file, vector_files.end());
+ ASSERT_EQ(vector_file->first, next_row_id);
+ next_row_id += vector_file->second->row_count;
+ ASSERT_LE(next_row_id, end);
+ ++vector_file;
+ }
+ }
+ ASSERT_EQ(next_row_id, input->length());
+ ASSERT_EQ(vector_file, vector_files.end());
+ ASSERT_OK(ScanAndRead(table_path, {"id", "embedding", "tag"}, input));
+ auto normal_projection =
+ arrow::StructArray::Make({input->field(0), input->field(2)}, {fields[0], fields[2]})
+ .ValueOrDie();
+ ASSERT_OK(ScanAndRead(table_path, {"id", "tag"}, normal_projection));
+ auto vector_projection = arrow::StructArray::Make({input->field(1)}, {fields[1]}).ValueOrDie();
+ ASSERT_OK(ScanAndRead(table_path, {"embedding"}, vector_projection));
+ for (const Range& range : {Range(999, 1001), Range(2999, 3001), Range(5999, 6000)}) {
+ auto expected = checked_pointer_cast<arrow::StructArray>(
+ input->Slice(range.from, range.to - range.from + 1));
+ ASSERT_OK(ScanAndRead(table_path, {"id", "embedding", "tag"}, expected,
+ /*predicate=*/nullptr, /*row_ranges=*/{range}));
+ }
+
+ ASSERT_OK_AND_ASSIGN(messages, WriteArray(table_path, {}, input->Slice(0, 2)));
+ ASSERT_OK(Commit(table_path, messages));
+ auto appended = checked_pointer_cast<arrow::StructArray>(input->Slice(0, 2));
+ ASSERT_OK(ScanAndRead(table_path, {"id", "embedding", "tag"}, appended,
+ /*predicate=*/nullptr, /*row_ranges=*/{Range(6001, 6002)}));
+}
+
TEST_P(DataEvolutionTableTest, TestVectorReadWrite) {
if (FileFormat() != "parquet" && FileFormat() != "lance") {
return;
@@ -3576,6 +4014,26 @@ std::vector<DataEvolutionTableParam> GetTestValuesForDataEvolutionDvTableTest()
INSTANTIATE_TEST_SUITE_P(FileFormat, DataEvolutionDvTableTest,
::testing::ValuesIn(GetTestValuesForDataEvolutionDvTableTest()));
+std::vector<DataEvolutionTableParam> GetTestValuesForDedicatedVectorTableTest() {
+ std::vector<std::string> formats = {"parquet"};
+#ifdef PAIMON_ENABLE_LANCE
+ formats.push_back("lance");
+#endif
+#ifdef PAIMON_ENABLE_VORTEX
+ formats.push_back("vortex");
+#endif
+ std::vector<DataEvolutionTableParam> values;
+ for (const auto& format : formats) {
+ for (bool enable_cache : {false, true}) {
+ values.emplace_back(format, enable_cache, /*bitmap64=*/false);
+ }
+ }
+ return values;
+}
+
+INSTANTIATE_TEST_SUITE_P(VectorFormat, DedicatedVectorTableTest,
+ ::testing::ValuesIn(GetTestValuesForDedicatedVectorTableTest()));
+
INSTANTIATE_TEST_SUITE_P(FileFormat, DataEvolutionTableTest,
::testing::ValuesIn(GetTestValuesForDataEvolutionTableTest()));
diff --git a/test/inte/write_and_read_inte_test.cpp b/test/inte/write_and_read_inte_test.cpp
index abff6ec7..441c6d81 100644
--- a/test/inte/write_and_read_inte_test.cpp
+++ b/test/inte/write_and_read_inte_test.cpp
@@ -5871,6 +5871,82 @@ TEST_P(WriteAndReadInteTest, TestSharedShreddingAllNullMapColumn) {
}
#ifdef PAIMON_ENABLE_VORTEX
+TEST(VortexVectorWriteInteTest, TestAppendVectorReadWrite) {
+ auto dir = UniqueTestDirectory::Create("local");
+ ASSERT_NE(dir, nullptr);
+ auto vector_type =
+ arrow::fixed_size_list(arrow::field("item", arrow::float32(), /*nullable=*/false), 3);
+ arrow::FieldVector fields = {
+ arrow::field("id", arrow::int32()), arrow::field("embedding", vector_type),
+ arrow::field("payload", arrow::struct_({arrow::field("embedding", vector_type)}))};
+ std::map<std::string, std::string> options = {
+ {Options::FILE_FORMAT, "vortex"}, {Options::BUCKET, "-1"}, {Options::FILE_SYSTEM, "local"}};
+ ASSERT_OK_AND_ASSIGN(
+ std::unique_ptr<TestHelper> helper,
+ TestHelper::Create(dir->Str(), arrow::schema(fields), /*partition_keys=*/{},
+ /*primary_keys=*/{}, options, /*is_streaming_mode=*/false));
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch,
+ TestHelper::MakeRecordBatch(arrow::struct_(fields), R"([
+ [1, [1, 2, 3], [[4, 5, 6]]],
+ [2, null, [null]],
+ [3, [7, 8, 9], null]
+ ])",
+ /*partition_map=*/{}, /*bucket=*/0, {}));
+ ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> commit_messages,
+ helper->WriteAndCommit(std::move(batch), /*commit_identifier=*/0,
+ /*expected_commit_messages=*/std::nullopt));
+ ASSERT_FALSE(commit_messages.empty());
+
+ arrow::FieldVector result_fields = fields;
+ result_fields.insert(result_fields.begin(), arrow::field("_VALUE_KIND", arrow::int8()));
+ ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<Split>> splits,
+ helper->NewScan(StartupMode::LatestFull(), /*snapshot_id=*/std::nullopt));
+ ASSERT_OK_AND_ASSIGN(bool success,
+ helper->ReadAndCheckResult(arrow::struct_(result_fields), splits, R"([
+ [0, 1, [1, 2, 3], [[4, 5, 6]]],
+ [0, 2, null, [null]],
+ [0, 3, [7, 8, 9], null]
+ ])"));
+ ASSERT_TRUE(success);
+}
+
+TEST(VortexVectorWriteInteTest, TestPrimaryKeyVectorUpdate) {
+ auto dir = UniqueTestDirectory::Create("local");
+ ASSERT_NE(dir, nullptr);
+ auto vector_type =
+ arrow::fixed_size_list(arrow::field("item", arrow::float32(), /*nullable=*/false), 3);
+ arrow::FieldVector fields = {arrow::field("pk", arrow::utf8()),
+ arrow::field("embedding", vector_type)};
+ std::map<std::string, std::string> options = {
+ {Options::FILE_FORMAT, "vortex"}, {Options::BUCKET, "1"}, {Options::FILE_SYSTEM, "local"}};
+ ASSERT_OK_AND_ASSIGN(
+ std::unique_ptr<TestHelper> helper,
+ TestHelper::Create(dir->Str(), arrow::schema(fields), /*partition_keys=*/{},
+ /*primary_keys=*/{"pk"}, options, /*is_streaming_mode=*/true));
+ const std::vector<std::string> batches = {R"([["a", [1, 2, 3]], ["b", [4, 5, 6]]])",
+ R"([["a", [7, 8, 9]], ["b", null]])"};
+ int64_t commit_identifier = 0;
+ for (const std::string& data_json : batches) {
+ ASSERT_OK_AND_ASSIGN(
+ std::unique_ptr<RecordBatch> batch,
+ TestHelper::MakeRecordBatch(arrow::struct_(fields), data_json, /*partition_map=*/{},
+ /*bucket=*/0, {}));
+ ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> commit_messages,
+ helper->WriteAndCommit(std::move(batch), commit_identifier++,
+ /*expected_commit_messages=*/std::nullopt));
+ ASSERT_FALSE(commit_messages.empty());
+ }
+
+ arrow::FieldVector result_fields = fields;
+ result_fields.insert(result_fields.begin(), arrow::field("_VALUE_KIND", arrow::int8()));
+ ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<Split>> splits,
+ helper->NewScan(StartupMode::LatestFull(), /*snapshot_id=*/std::nullopt));
+ ASSERT_OK_AND_ASSIGN(bool success,
+ helper->ReadAndCheckResult(arrow::struct_(result_fields), splits,
+ R"([[0, "a", [7, 8, 9]], [0, "b", null]])"));
+ ASSERT_TRUE(success);
+}
+
// Regression: primary-key finalization requires one stats entry per write-schema field
// (KeyValueDataFileWriter), which the Vortex extractor satisfies with unknown entries.
TEST(VortexPrimaryKeyWriteInteTest, TestPKWriteFinalize) {
diff --git a/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/README.md b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/README.md
new file mode 100644
index 00000000..bd16b1dc
--- /dev/null
+++ b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/README.md
@@ -0,0 +1,39 @@
+Table: dedicated_vector_java_compat
+Writer: Paimon Java 2.2-SNAPSHOT (vortex-jni 0.73.0), JDK 17
+Java source commit: 7d6d146bbaf33a349dc83c3caf71334bd551e3d5
+
+Schema (all columns are nullable):
+id INT
+embedding VECTOR<FLOAT, 3>
+tag STRING
+
+No partition key or primary key.
+
+Options:
+file.format = parquet
+vector.file.format = vortex
+bucket = -1
+row-tracking.enabled = true
+data-evolution.enabled = true
+write-only = true
+
+Msgs:
+snapshot-1 APPEND
+Add: (1, [1, 2, 3], "a")
+Add: (2, NULL, "b")
+Add: (3, [-1, 0, 1], "c")
+
+snapshot-2 APPEND
+Add: (4, NULL, "d")
+Add: (5, [7, 8, 9], "e")
+Add: (6, [10, 11, 12], "f")
+
+Each snapshot adds one Parquet main file (writeCols=[id, tag]) and one Vortex
+VECTOR file (writeCols=[embedding]). Each file has three rows. The first pair
+covers row ids [0, 2]; the second pair covers [3, 5].
+
+Generated using the Paimon Java batch write API (BatchWriteBuilder /
+BatchTableWrite / BatchTableCommit). Create the table with the schema and
+options above, then write and commit each three-row batch separately.
+write-only prevents compaction, preserving both append commits. Java readback
+verified every row and file range before copying the table files unchanged.
diff --git a/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/bucket-0/data-02c40850-170e-4085-bc42-a69c6d6fffdc-0.parquet b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/bucket-0/data-02c40850-170e-4085-bc42-a69c6d6fffdc-0.parquet
new file mode 100644
index 00000000..2a1a3aac
Binary files /dev/null and b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/bucket-0/data-02c40850-170e-4085-bc42-a69c6d6fffdc-0.parquet differ
diff --git a/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/bucket-0/data-02c40850-170e-4085-bc42-a69c6d6fffdc-1.vector.vortex b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/bucket-0/data-02c40850-170e-4085-bc42-a69c6d6fffdc-1.vector.vortex
new file mode 100644
index 00000000..b52459a4
Binary files /dev/null and b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/bucket-0/data-02c40850-170e-4085-bc42-a69c6d6fffdc-1.vector.vortex differ
diff --git a/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/bucket-0/data-bd5ba451-e0e6-4a75-8a36-2ea3d2387dbe-0.parquet b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/bucket-0/data-bd5ba451-e0e6-4a75-8a36-2ea3d2387dbe-0.parquet
new file mode 100644
index 00000000..4d5d1d34
Binary files /dev/null and b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/bucket-0/data-bd5ba451-e0e6-4a75-8a36-2ea3d2387dbe-0.parquet differ
diff --git a/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/bucket-0/data-bd5ba451-e0e6-4a75-8a36-2ea3d2387dbe-1.vector.vortex b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/bucket-0/data-bd5ba451-e0e6-4a75-8a36-2ea3d2387dbe-1.vector.vortex
new file mode 100644
index 00000000..a311aeff
Binary files /dev/null and b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/bucket-0/data-bd5ba451-e0e6-4a75-8a36-2ea3d2387dbe-1.vector.vortex differ
diff --git a/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-16ae6a76-0eec-45b3-a1e2-f18249238804-0 b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-16ae6a76-0eec-45b3-a1e2-f18249238804-0
new file mode 100644
index 00000000..48fabbdb
Binary files /dev/null and b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-16ae6a76-0eec-45b3-a1e2-f18249238804-0 differ
diff --git a/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-736f9660-1f51-41e2-b0bf-751a67514c1d-0 b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-736f9660-1f51-41e2-b0bf-751a67514c1d-0
new file mode 100644
index 00000000..f7a3604d
Binary files /dev/null and b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-736f9660-1f51-41e2-b0bf-751a67514c1d-0 differ
diff --git a/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-list-d1d44abf-1ad2-4f22-bd0d-bc8b0d1ab444-0 b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-list-d1d44abf-1ad2-4f22-bd0d-bc8b0d1ab444-0
new file mode 100644
index 00000000..4896c71a
Binary files /dev/null and b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-list-d1d44abf-1ad2-4f22-bd0d-bc8b0d1ab444-0 differ
diff --git a/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-list-d1d44abf-1ad2-4f22-bd0d-bc8b0d1ab444-1 b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-list-d1d44abf-1ad2-4f22-bd0d-bc8b0d1ab444-1
new file mode 100644
index 00000000..cc700356
Binary files /dev/null and b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-list-d1d44abf-1ad2-4f22-bd0d-bc8b0d1ab444-1 differ
diff --git a/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-list-d7544098-f6d3-4754-8d1c-cc9e672f1970-0 b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-list-d7544098-f6d3-4754-8d1c-cc9e672f1970-0
new file mode 100644
index 00000000..122e2d8c
Binary files /dev/null and b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-list-d7544098-f6d3-4754-8d1c-cc9e672f1970-0 differ
diff --git a/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-list-d7544098-f6d3-4754-8d1c-cc9e672f1970-1 b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-list-d7544098-f6d3-4754-8d1c-cc9e672f1970-1
new file mode 100644
index 00000000..d834da2e
Binary files /dev/null and b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-list-d7544098-f6d3-4754-8d1c-cc9e672f1970-1 differ
diff --git a/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/schema/schema-0 b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/schema/schema-0
new file mode 100644
index 00000000..4a7d5c83
--- /dev/null
+++ b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/schema/schema-0
@@ -0,0 +1,33 @@
+{
+ "version" : 3,
+ "id" : 0,
+ "fields" : [ {
+ "id" : 0,
+ "name" : "id",
+ "type" : "INT"
+ }, {
+ "id" : 1,
+ "name" : "embedding",
+ "type" : {
+ "type" : "VECTOR",
+ "element" : "FLOAT",
+ "length" : 3
+ }
+ }, {
+ "id" : 2,
+ "name" : "tag",
+ "type" : "STRING"
+ } ],
+ "highestFieldId" : 2,
+ "partitionKeys" : [ ],
+ "primaryKeys" : [ ],
+ "options" : {
+ "bucket" : "-1",
+ "row-tracking.enabled" : "true",
+ "vector.file.format" : "vortex",
+ "data-evolution.enabled" : "true",
+ "write-only" : "true",
+ "file.format" : "parquet"
+ },
+ "timeMillis" : 1792119586027
+}
\ No newline at end of file
diff --git a/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/snapshot/LATEST b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/snapshot/LATEST
new file mode 100644
index 00000000..d8263ee9
--- /dev/null
+++ b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/snapshot/LATEST
@@ -0,0 +1 @@
+2
\ No newline at end of file
diff --git a/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/snapshot/snapshot-1 b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/snapshot/snapshot-1
new file mode 100644
index 00000000..13f94365
--- /dev/null
+++ b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/snapshot/snapshot-1
@@ -0,0 +1,18 @@
+{
+ "version" : 3,
+ "uuid" : "5ef42272-b719-4159-aa15-87450d35ab82",
+ "id" : 1,
+ "schemaId" : 0,
+ "baseManifestList" : "manifest-list-d7544098-f6d3-4754-8d1c-cc9e672f1970-0",
+ "baseManifestListSize" : 1158,
+ "deltaManifestList" : "manifest-list-d7544098-f6d3-4754-8d1c-cc9e672f1970-1",
+ "deltaManifestListSize" : 1261,
+ "commitUser" : "99aaaab3-b662-4b9e-b047-d61d727d77d6",
+ "writerVersion" : "java-2.2-SNAPSHOT-7d6d146bbaf33a349dc83c3caf71334bd551e3d5",
+ "commitIdentifier" : 9223372036854775807,
+ "commitKind" : "APPEND",
+ "timeMillis" : 1792119587829,
+ "totalRecordCount" : 6,
+ "deltaRecordCount" : 6,
+ "nextRowId" : 3
+}
\ No newline at end of file
diff --git a/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/snapshot/snapshot-2 b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/snapshot/snapshot-2
new file mode 100644
index 00000000..f3767d9f
--- /dev/null
+++ b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/snapshot/snapshot-2
@@ -0,0 +1,18 @@
+{
+ "version" : 3,
+ "uuid" : "162bab7a-d0d3-4277-945c-ce4818626d26",
+ "id" : 2,
+ "schemaId" : 0,
+ "baseManifestList" : "manifest-list-d1d44abf-1ad2-4f22-bd0d-bc8b0d1ab444-0",
+ "baseManifestListSize" : 1261,
+ "deltaManifestList" : "manifest-list-d1d44abf-1ad2-4f22-bd0d-bc8b0d1ab444-1",
+ "deltaManifestListSize" : 1263,
+ "commitUser" : "a4e92eca-e481-42ac-a96e-f44f7ba79239",
+ "writerVersion" : "java-2.2-SNAPSHOT-7d6d146bbaf33a349dc83c3caf71334bd551e3d5",
+ "commitIdentifier" : 9223372036854775807,
+ "commitKind" : "APPEND",
+ "timeMillis" : 1792119587887,
+ "totalRecordCount" : 12,
+ "deltaRecordCount" : 6,
+ "nextRowId" : 6
+}
\ No newline at end of file
diff --git a/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/README.md b/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/README.md
new file mode 100644
index 00000000..4ebc9e56
--- /dev/null
+++ b/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/README.md
@@ -0,0 +1,46 @@
+Table: dedicated_vector_java_rolling
+Writer: Paimon Java 2.2-SNAPSHOT (vortex-jni 0.73.0), JDK 17
+Java source commit: 7d6d146bbaf33a349dc83c3caf71334bd551e3d5
+
+Schema (all columns are nullable):
+id INT
+embedding VECTOR<FLOAT, 3>
+tag STRING
+
+No partition key or primary key.
+
+Options:
+file.format = parquet
+vector.file.format = vortex
+bucket = -1
+row-tracking.enabled = true
+data-evolution.enabled = true
+write-only = true
+target-file-size = 128 mb
+vector.target-file-size = 1 kb
+write.batch-size = 100
+read.batch-size = 127
+
+Msgs:
+snapshot-1 APPEND
+Add 3001 rows in ascending id order, from 0 through 3000:
+(id, [id, id + 1, id + 2], "row-<id>")
+VECTOR elements are FLOAT values. embedding is NULL at ids
+0, 999, 1000, 1999, 2000, 2999, and 3000.
+
+Files (inclusive row id ranges):
+Parquet main: [0, 3000], 3001 rows, writeCols=[id, tag]
+Vortex VECTOR: [0, 999], 1000 rows, writeCols=[embedding]
+Vortex VECTOR: [1000, 1999], 1000 rows, writeCols=[embedding]
+Vortex VECTOR: [2000, 2999], 1000 rows, writeCols=[embedding]
+Vortex VECTOR: [3000, 3000], 1 row, writeCols=[embedding]
+
+Generated using the Paimon Java batch write API (BatchWriteBuilder /
+BatchTableWrite / BatchTableCommit). Create the table with the schema and
+options above, write all 3001 rows, then prepare and commit once.
+write-only prevents compaction. The small write batch flushes Vortex buffers
+before Java checks rolling every 1000 rows. The VECTOR files roll while the
+main file stays open; vector.target-file-size is a target, not a physical size
+limit. NULL vectors cover both sides of file boundaries and the final tail;
+read.batch-size=127 also exercises reads across those boundaries. Java readback
+verified every row and file range before copying the table files unchanged.
diff --git a/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/bucket-0/data-253f5756-e2a8-4ab4-92f0-c0896ef10d06-0.parquet b/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/bucket-0/data-253f5756-e2a8-4ab4-92f0-c0896ef10d06-0.parquet
new file mode 100644
index 00000000..fa539358
Binary files /dev/null and b/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/bucket-0/data-253f5756-e2a8-4ab4-92f0-c0896ef10d06-0.parquet differ
diff --git a/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/bucket-0/data-253f5756-e2a8-4ab4-92f0-c0896ef10d06-1.vector.vortex b/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/bucket-0/data-253f5756-e2a8-4ab4-92f0-c0896ef10d06-1.vector.vortex
new file mode 100644
index 00000000..bcddf61a
Binary files /dev/null and b/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/bucket-0/data-253f5756-e2a8-4ab4-92f0-c0896ef10d06-1.vector.vortex differ
diff --git a/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/bucket-0/data-253f5756-e2a8-4ab4-92f0-c0896ef10d06-2.vector.vortex b/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/bucket-0/data-253f5756-e2a8-4ab4-92f0-c0896ef10d06-2.vector.vortex
new file mode 100644
index 00000000..018d93df
Binary files /dev/null and b/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/bucket-0/data-253f5756-e2a8-4ab4-92f0-c0896ef10d06-2.vector.vortex differ
diff --git a/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/bucket-0/data-253f5756-e2a8-4ab4-92f0-c0896ef10d06-3.vector.vortex b/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/bucket-0/data-253f5756-e2a8-4ab4-92f0-c0896ef10d06-3.vector.vortex
new file mode 100644
index 00000000..dff6e49c
Binary files /dev/null and b/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/bucket-0/data-253f5756-e2a8-4ab4-92f0-c0896ef10d06-3.vector.vortex differ
diff --git a/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/bucket-0/data-253f5756-e2a8-4ab4-92f0-c0896ef10d06-4.vector.vortex b/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/bucket-0/data-253f5756-e2a8-4ab4-92f0-c0896ef10d06-4.vector.vortex
new file mode 100644
index 00000000..4f238d1b
Binary files /dev/null and b/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/bucket-0/data-253f5756-e2a8-4ab4-92f0-c0896ef10d06-4.vector.vortex differ
diff --git a/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/manifest/manifest-af194a04-83f3-4b43-a88b-10f8e3d36dd9-0 b/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/manifest/manifest-af194a04-83f3-4b43-a88b-10f8e3d36dd9-0
new file mode 100644
index 00000000..ff65bf02
Binary files /dev/null and b/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/manifest/manifest-af194a04-83f3-4b43-a88b-10f8e3d36dd9-0 differ
diff --git a/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/manifest/manifest-list-947870e6-2276-4657-a9fc-ac454ea83ffd-0 b/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/manifest/manifest-list-947870e6-2276-4657-a9fc-ac454ea83ffd-0
new file mode 100644
index 00000000..e73240fe
Binary files /dev/null and b/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/manifest/manifest-list-947870e6-2276-4657-a9fc-ac454ea83ffd-0 differ
diff --git a/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/manifest/manifest-list-947870e6-2276-4657-a9fc-ac454ea83ffd-1 b/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/manifest/manifest-list-947870e6-2276-4657-a9fc-ac454ea83ffd-1
new file mode 100644
index 00000000..2003989c
Binary files /dev/null and b/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/manifest/manifest-list-947870e6-2276-4657-a9fc-ac454ea83ffd-1 differ
diff --git a/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/schema/schema-0 b/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/schema/schema-0
new file mode 100644
index 00000000..1a9df8cc
--- /dev/null
+++ b/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/schema/schema-0
@@ -0,0 +1,37 @@
+{
+ "version" : 3,
+ "id" : 0,
+ "fields" : [ {
+ "id" : 0,
+ "name" : "id",
+ "type" : "INT"
+ }, {
+ "id" : 1,
+ "name" : "embedding",
+ "type" : {
+ "type" : "VECTOR",
+ "element" : "FLOAT",
+ "length" : 3
+ }
+ }, {
+ "id" : 2,
+ "name" : "tag",
+ "type" : "STRING"
+ } ],
+ "highestFieldId" : 2,
+ "partitionKeys" : [ ],
+ "primaryKeys" : [ ],
+ "options" : {
+ "bucket" : "-1",
+ "read.batch-size" : "127",
+ "row-tracking.enabled" : "true",
+ "vector.file.format" : "vortex",
+ "target-file-size" : "128 mb",
+ "data-evolution.enabled" : "true",
+ "write-only" : "true",
+ "vector.target-file-size" : "1 kb",
+ "file.format" : "parquet",
+ "write.batch-size" : "100"
+ },
+ "timeMillis" : 1792376046708
+}
\ No newline at end of file
diff --git a/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/snapshot/LATEST b/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/snapshot/LATEST
new file mode 100644
index 00000000..56a6051c
--- /dev/null
+++ b/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/snapshot/LATEST
@@ -0,0 +1 @@
+1
\ No newline at end of file
diff --git a/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/snapshot/snapshot-1 b/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/snapshot/snapshot-1
new file mode 100644
index 00000000..64a54b13
--- /dev/null
+++ b/test/test_data/vortex/dedicated_vector_java_rolling.db/dedicated_vector_java_rolling/snapshot/snapshot-1
@@ -0,0 +1,18 @@
+{
+ "version" : 3,
+ "uuid" : "681d6dac-f32d-4d38-a28c-e250dcfc6f82",
+ "id" : 1,
+ "schemaId" : 0,
+ "baseManifestList" : "manifest-list-947870e6-2276-4657-a9fc-ac454ea83ffd-0",
+ "baseManifestListSize" : 1158,
+ "deltaManifestList" : "manifest-list-947870e6-2276-4657-a9fc-ac454ea83ffd-1",
+ "deltaManifestListSize" : 1262,
+ "commitUser" : "0e4cea46-6429-4b78-9c82-69df4b08d96b",
+ "writerVersion" : "java-2.2-SNAPSHOT-7d6d146bbaf33a349dc83c3caf71334bd551e3d5",
+ "commitIdentifier" : 9223372036854775807,
+ "commitKind" : "APPEND",
+ "timeMillis" : 1792376049116,
+ "totalRecordCount" : 6002,
+ "deltaRecordCount" : 6002,
+ "nextRowId" : 3001
+}
\ No newline at end of file