feat(vector): add dedicated Vortex storage for append-only tables

docs/source/user_guide/data_types.rst

@@ -197,13 +197,26 @@ 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.
+
+       With Vortex enabled at build time, ``vector.file.format=vortex`` stores a
+       single top-level VECTOR column in dedicated ``.vector.vortex`` files.
+       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.
+
+       Main and VECTOR files cover identical row ranges and roll together when
+       either size target or ``target-file-row-num`` is reached. Configure the
+       VECTOR size target with ``vector.target-file-size``; it defaults to
+       ``target-file-size``. Independent VECTOR file rolling is not supported.
 
        Paimon C++ also reads Parquet files written by Paimon Rust or Python whose
        embedded Arrow schema restores VECTOR columns as ``FixedSizeList``,

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. The initial implementation
+    /// supports 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.
+    /// Main and vector files currently roll together when either target is reached.
+    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[];

src/paimon/CMakeLists.txt

@@ -316,6 +316,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
@@ -833,6 +835,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

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";

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()

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;
@@ -598,6 +600,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
@@ -1119,6 +1130,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;
 }
@@ -1165,6 +1180,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

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;

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);

src/paimon/core/io/rolling_vector_file_writer.cpp

@@ -0,0 +1,158 @@
+/*
+ * 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());
+        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 || vector_full || current_file_record_count_ >= target_file_row_num_) {
+        PAIMON_RETURN_NOT_OK(CloseCurrentPair());
+    }
+    guard.Release();
+    return Status::OK();
+}
+
+Status RollingVectorFileWriter::CloseCurrentPair() {
+    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(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> main_meta, current_writer_->GetResult());
+    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<DataFileMeta> vector_meta, vector_writer_->GetResult());
+    if (main_meta->row_count != vector_meta->row_count) {
+        return Status::Invalid("Main and VECTOR files must contain the same number of rows");
+    }
+    // Row tracking assigns ids in manifest order: the normal anchor must precede its vector file.
+    results_.push_back(main_meta);
+    results_.push_back(vector_meta);
+    metrics_->Merge(current_writer_->GetMetrics());
+    metrics_->Merge(vector_writer_->GetMetrics());
+    current_writer_.reset();
+    vector_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 = CloseCurrentPair();
+    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();
+    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