From dff326640eda3c82c8db66cf33d30ecaf42f19f5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=B0=8F=E6=98=8E=E5=90=8C=E5=AD=A6?= <72908278+ChaomingZhangCN@users.noreply.github.com> Date: Fri, 16 Oct 2026 10:24:47 +0800 Subject: [PATCH 1/2] feat(vector): add dedicated Vortex storage for append-only tables --- docs/source/user_guide/data_types.rst | 21 +- 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/io/data_file_path_factory.h | 4 + .../core/io/rolling_vector_file_writer.cpp | 158 +++++++++++++++ .../core/io/rolling_vector_file_writer.h | 56 ++++++ .../io/rolling_vector_file_writer_test.cpp | 189 ++++++++++++++++++ .../io/vector_data_file_writer_factory.cpp | 49 +++++ .../core/io/vector_data_file_writer_factory.h | 34 ++++ .../core/operation/file_store_write.cpp | 6 + src/paimon/core/schema/schema_validation.cpp | 62 +++++- src/paimon/core/schema/schema_validation.h | 3 + .../core/schema/schema_validation_test.cpp | 87 +++++++- .../format/vortex/vortex_file_format_test.cpp | 47 +++++ .../format/vortex/vortex_format_writer.cpp | 38 ++++ test/inte/data_evolution_table_test.cpp | 102 ++++++++++ test/inte/write_and_read_inte_test.cpp | 76 +++++++ 21 files changed, 988 insertions(+), 10 deletions(-) create mode 100644 src/paimon/core/io/rolling_vector_file_writer.cpp create mode 100644 src/paimon/core/io/rolling_vector_file_writer.h create mode 100644 src/paimon/core/io/rolling_vector_file_writer_test.cpp create mode 100644 src/paimon/core/io/vector_data_file_writer_factory.cpp create mode 100644 src/paimon/core/io/vector_data_file_writer_factory.h diff --git a/docs/source/user_guide/data_types.rst b/docs/source/user_guide/data_types.rst index f864596df..16ad190e1 100644 --- a/docs/source/user_guide/data_types.rst +++ b/docs/source/user_guide/data_types.rst @@ -197,13 +197,26 @@ and `Arrow DataTypes 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( + options_, schema_id_, vector_schema, vector_schema->field_names(), seq_num_counter_, + FileSource::Append(), path_factory_, memory_pool_); + return std::make_unique( + 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 28af19140..443a9077e 100644 --- a/src/paimon/core/core_options.cpp +++ b/src/paimon/core/core_options.cpp @@ -358,7 +358,9 @@ struct CoreOptions::Impl { std::optional target_file_size; std::optional blob_target_file_size; + std::optional vector_target_file_size; std::shared_ptr file_format; + std::shared_ptr vector_file_format; std::shared_ptr file_system; std::shared_ptr manifest_file_format; std::shared_ptr cache; @@ -598,6 +600,15 @@ struct CoreOptions::Impl { // Parse file.format - data file format, default "parquet" PAIMON_RETURN_NOT_OK(parser.ParseObject( Options::FILE_FORMAT, /*default_identifier=*/"parquet", &file_format)); + if (parser.ContainsKey(Options::VECTOR_FILE_FORMAT)) { + PAIMON_RETURN_NOT_OK(parser.ParseObject( + 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 CoreOptions::GetFileFormat() const { return impl_->file_format; } +std::shared_ptr CoreOptions::GetVectorFileFormat() const { + return impl_->vector_file_format; +} + std::shared_ptr 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 diff --git a/src/paimon/core/core_options.h b/src/paimon/core/core_options.h index a16f75e4c..62b611612 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 GetFileFormat() const; + std::shared_ptr GetVectorFileFormat() const; std::shared_ptr GetWriteFileFormat(int32_t level) const; std::shared_ptr 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/io/data_file_path_factory.h b/src/paimon/core/io/data_file_path_factory.h index a9050eeee..54f4168ef 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 000000000..3f647be6f --- /dev/null +++ b/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& main_factory, + const std::shared_ptr& vector_factory, + const std::shared_ptr& 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 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(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 main_array, + arrow::StructArray::Make(main_arrays, main_fields)); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + std::shared_ptr 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 main_meta, current_writer_->GetResult()); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr 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>> 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 000000000..c8eee9126 --- /dev/null +++ b/src/paimon/core/io/rolling_vector_file_writer.h @@ -0,0 +1,56 @@ +/* + * 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 to files covering identical row ranges. +class RollingVectorFileWriter + : public RollingFileWriter<::ArrowArray*, std::shared_ptr> { + public: + using Writer = SingleFileWriter<::ArrowArray*, std::shared_ptr>; + using WriterFactory = SingleFileWriterFactory<::ArrowArray*, std::shared_ptr>; + + RollingVectorFileWriter(int64_t target_file_size, int64_t target_file_row_num, + int64_t vector_target_file_size, + const std::shared_ptr& main_factory, + const std::shared_ptr& vector_factory, + const std::shared_ptr& write_schema, + int32_t vector_index); + + Status Write(::ArrowArray* record) override; + Status Close() override; + void Abort() override; + Result>> GetResult() override; + + private: + Status CloseCurrentPair(); + + int64_t vector_target_file_size_; + std::shared_ptr vector_factory_; + std::shared_ptr write_schema_; + int32_t vector_index_; + std::unique_ptr vector_writer_; + 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 000000000..51672f8b3 --- /dev/null +++ b/src/paimon/core/io/rolling_vector_file_writer_test.cpp @@ -0,0 +1,189 @@ +/* + * 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 +#include + +#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; + int64_t file_number = 0; + std::vector paths; +}; + +class TestVectorWriter : public RollingVectorFileWriter::Writer { + public: + TestVectorWriter(const std::shared_ptr& fs, const std::string& path, + const std::shared_ptr& 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 ReachTargetSize(bool, int64_t target_size) override { + return rows_ >= target_size; + } + + Result> GetResult() override { + return DataFileMeta::ForAppend( + PathUtil::GetName(path_), 0, rows_, SimpleStats::EmptyStats(), 0, rows_ - 1, 0, + FileSource::Append(), std::nullopt, std::nullopt, std::nullopt, std::nullopt); + } + + private: + std::shared_ptr state_; + int64_t rows_ = 0; +}; + +class TestVectorWriterFactory : public RollingVectorFileWriter::WriterFactory { + public: + TestVectorWriterFactory(std::shared_ptr fs, std::string prefix, + std::shared_ptr state) + : fs_(std::move(fs)), prefix_(std::move(prefix)), state_(std::move(state)) {} + + Result> 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 stream, + fs_->Create(path, /*overwrite=*/false)); + PAIMON_RETURN_NOT_OK(stream->Close()); + state_->paths.push_back(path); + return std::make_unique(fs_, path, state_); + } + + private: + std::shared_ptr fs_; + std::string prefix_; + std::shared_ptr state_; +}; + +} // namespace + +TEST(RollingVectorFileWriterTest, VectorTargetRollsBothFiles) { + auto dir = UniqueTestDirectory::Create("local"); + ASSERT_NE(dir, nullptr); + std::shared_ptr fs = dir->GetFileSystem(); + auto main_state = std::make_shared(); + auto vector_state = std::make_shared(); + auto main_factory = std::make_shared( + fs, PathUtil::JoinPath(dir->Str(), "main"), main_state); + auto vector_factory = std::make_shared( + fs, PathUtil::JoinPath(dir->Str(), "vector"), vector_state); + auto schema = arrow::schema({arrow::field("id", arrow::int32()), + arrow::field("v", arrow::fixed_size_list(arrow::float32(), 2))}); + RollingVectorFileWriter writer(1024, std::numeric_limits::max(), 1, main_factory, + vector_factory, schema, 1); + auto array = arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(schema->fields()), + R"([[1, [1, 2]], [2, null]])") + .ValueOrDie(); + for (int32_t i = 0; i < 2; ++i) { + ::ArrowArray batch = {}; + ASSERT_TRUE(arrow::ExportArray(*array, &batch).ok()); + ASSERT_OK(writer.Write(&batch)); + } + ASSERT_OK(writer.Close()); + ASSERT_OK(writer.Close()); + ASSERT_OK_AND_ASSIGN(std::vector> results, writer.GetResult()); + ASSERT_EQ(results.size(), 4); + for (const auto& file : results) { + ASSERT_EQ(file->row_count, 2); + } + writer.Abort(); + for (const auto& state : {main_state, vector_state}) { + ASSERT_EQ(state->paths.size(), 2); + for (const auto& path : state->paths) { + ASSERT_OK_AND_ASSIGN(bool exists, fs->Exists(path)); + ASSERT_FALSE(exists); + } + } + ASSERT_NOK(writer.GetResult()); +} + +TEST(RollingVectorFileWriterTest, FailureRemovesBothFilesAndEarlierPairs) { + for (int32_t failure = 0; failure < 6; ++failure) { + SCOPED_TRACE(failure); + auto dir = UniqueTestDirectory::Create("local"); + ASSERT_NE(dir, nullptr); + std::shared_ptr fs = dir->GetFileSystem(); + auto main_state = std::make_shared(); + auto vector_state = std::make_shared(); + auto main_factory = std::make_shared( + fs, PathUtil::JoinPath(dir->Str(), "main"), main_state); + auto vector_factory = std::make_shared( + fs, PathUtil::JoinPath(dir->Str(), "vector"), vector_state); + auto schema = + arrow::schema({arrow::field("id", arrow::int32()), + arrow::field("v", arrow::fixed_size_list(arrow::float32(), 2))}); + RollingVectorFileWriter writer(1024, 1, 1024, main_factory, vector_factory, schema, 1); + auto array = arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(schema->fields()), + R"([[1, [1, 2]]])") + .ValueOrDie(); + ::ArrowArray batch = {}; + ASSERT_TRUE(arrow::ExportArray(*array, &batch).ok()); + ASSERT_OK(writer.Write(&batch)); + const auto& failing_state = failure < 3 ? main_state : vector_state; + failing_state->fail_create = failure % 3 == 0; + failing_state->fail_write = failure % 3 == 1; + failing_state->fail_close = failure % 3 == 2; + ASSERT_TRUE(arrow::ExportArray(*array, &batch).ok()); + ASSERT_NOK(writer.Write(&batch)); + ASSERT_NOK(writer.Close()); + ASSERT_NOK(writer.GetResult()); + 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); + } + } + } +} + +} // 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 000000000..e2812bd7f --- /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 + +#include "paimon/core/io/data_file_path_factory.h" +#include "paimon/format/file_format.h" + +namespace paimon { + +Result>>> +VectorDataFileWriterFactory::CreateWriter() const { + const std::shared_ptr 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( + options_.GetFileCompression(), std::function(), + 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>>( + 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 000000000..2f5be176c --- /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>>> + CreateWriter() const override; +}; + +} // namespace paimon diff --git a/src/paimon/core/operation/file_store_write.cpp b/src/paimon/core/operation/file_store_write.cpp index 2ae664f8c..3907c5384 100644 --- a/src/paimon/core/operation/file_store_write.cpp +++ b/src/paimon/core/operation/file_store_write.cpp @@ -49,6 +49,7 @@ #include "paimon/core/realtime/realtime_context_impl.h" #include "paimon/core/realtime/realtime_schema_layout.h" #include "paimon/core/schema/schema_manager.h" +#include "paimon/core/schema/schema_validation.h" #include "paimon/core/schema/table_schema.h" #include "paimon/core/table/bucket_mode.h" #include "paimon/core/table/format/format_table_file_store_write.h" @@ -248,6 +249,11 @@ Result> FileStoreWrite::Create(std::unique_ptrGetFileSystemSchemeToIdentifierMap())); + 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 partition_schema, FieldMapping::GetPartitionSchema(arrow_schema, schema->PartitionKeys())); diff --git a/src/paimon/core/schema/schema_validation.cpp b/src/paimon/core/schema/schema_validation.cpp index 24c2649fb..df36aa6c7 100644 --- a/src/paimon/core/schema/schema_validation.cpp +++ b/src/paimon/core/schema/schema_validation.cpp @@ -119,10 +119,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(); } @@ -391,6 +392,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(); } @@ -1212,6 +1214,58 @@ Status SchemaValidation::ValidateMapStorageLayout(const TableSchema& schema, return Status::OK(); } +Status SchemaValidation::ValidateVectorStore(const TableSchema& schema, + const CoreOptions& options) { + const std::shared_ptr vector_format = options.GetVectorFileFormat(); + if (!vector_format) { + return Status::OK(); + } + if (vector_format->Identifier() != "vortex") { + return Status::NotImplemented("Dedicated VECTOR storage currently only supports vortex"); + } + 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 c89b80197..cdee2e199 100644 --- a/src/paimon/core/schema/schema_validation.h +++ b/src/paimon/core/schema/schema_validation.h @@ -51,6 +51,9 @@ class SchemaValidation { static Status ValidateTableSchema(const TableSchema& schema); + /// 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 9356b0be1..d27e4b11b 100644 --- a/src/paimon/core/schema/schema_validation_test.cpp +++ b/src/paimon/core/schema/schema_validation_test.cpp @@ -128,7 +128,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 changelog_options = { {Options::BUCKET, "1"}, @@ -229,7 +229,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 @@ -407,6 +407,89 @@ TEST(SchemaValidationTest, TestLanceDataTypes) { #endif #ifdef PAIMON_ENABLE_VORTEX +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 options = {{Options::BUCKET, "-1"}, + {Options::FILE_FORMAT, "parquet"}, + {Options::VECTOR_FILE_FORMAT, "vortex"}, + {Options::ROW_TRACKING_ENABLED, "true"}, + {Options::DATA_EVOLUTION_ENABLED, "true"}}; + auto validate = [&](const arrow::FieldVector& fields, + const std::map& opts) -> Status { + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr 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 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"); + 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"); + invalid_options = options; + invalid_options[Options::VECTOR_FILE_FORMAT] = "parquet"; + ASSERT_NOK_WITH_MSG(validate({id, embedding}, invalid_options), "only supports 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); +} + +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>{ + {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 options = {{Options::BUCKET, "1"}, + {Options::FILE_FORMAT, "parquet"}}; + options[option_key] = option_value; + ASSERT_OK_AND_ASSIGN(std::shared_ptr 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 table_schema; // Vortex as the default file format: unsupported types are rejected at validation time. diff --git a/src/paimon/format/vortex/vortex_file_format_test.cpp b/src/paimon/format/vortex/vortex_file_format_test.cpp index 59bb7189f..dce351f69 100644 --- a/src/paimon/format/vortex/vortex_file_format_test.cpp +++ b/src/paimon/format/vortex/vortex_file_format_test.cpp @@ -450,6 +450,53 @@ TEST_F(VortexFileFormatTest, WriteThenReadFixedSizeListOfString) { AssertReadWithBatchSizes(path, schema, expected, {1, 2, 3, 4}); } +TEST_F(VortexFileFormatTest, WriteThenReadVectors) { + const std::vector> 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 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"); +} + // 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>. TEST_F(VortexFileFormatTest, WriteThenReadListOfStruct) { diff --git a/src/paimon/format/vortex/vortex_format_writer.cpp b/src/paimon/format/vortex/vortex_format_writer.cpp index 47d271118..e265c593d 100644 --- a/src/paimon/format/vortex/vortex_format_writer.cpp +++ b/src/paimon/format/vortex/vortex_format_writer.cpp @@ -21,18 +21,52 @@ #include +#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 "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/core/schema/schema_validation.h" #include "paimon/format/vortex/vortex_ffi_util.h" #include "paimon/fs/file_system.h" namespace paimon::vortex { +namespace { + +Result> NormalizeVectorNullElements( + const std::shared_ptr& data, arrow::compute::ExecContext* context) { + std::shared_ptr normalized = data; + for (size_t i = 0; i < data->child_data.size(); ++i) { + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr 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 may own null element slots. Vortex imports these masked slots too. + PAIMON_RETURN_NOT_OK(VectorUtils::ValidateVectorElements(*arrow::MakeArray(data))); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr 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 output, std::shared_ptr schema, VxSessionPtr session, @@ -103,6 +137,10 @@ Status VortexFormatWriter::AddBatch(::ArrowArray* batch) { arrow::ImportArray(batch, arrow::struct_(schema_->fields()))); PAIMON_ASSIGN_OR_RAISE(std::shared_ptr normalized, ArrowUtils::NormalizeArrayOffsets(imported, arrow_pool_.get())); + arrow::compute::ExecContext context(arrow_pool_.get()); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr normalized_data, + NormalizeVectorNullElements(normalized->data(), &context)); + normalized = arrow::MakeArray(normalized_data); 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. diff --git a/test/inte/data_evolution_table_test.cpp b/test/inte/data_evolution_table_test.cpp index 804e7c268..5939e3d39 100644 --- a/test/inte/data_evolution_table_test.cpp +++ b/test/inte/data_evolution_table_test.cpp @@ -30,6 +30,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 +1797,107 @@ TEST_P(DataEvolutionTableTest, TestPartitionWithPredicate) { } } +#ifdef PAIMON_ENABLE_VORTEX +TEST_P(DataEvolutionTableTest, TestDedicatedVectorStore) { + if (FileFormat() != "parquet") { + GTEST_SKIP() << "Dedicated VECTOR storage requires parquet main files"; + } + 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 options = { + {Options::FILE_FORMAT, "parquet"}, {Options::VECTOR_FILE_FORMAT, "vortex"}, + {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::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> batches = {input->Slice(1, 2), input->Slice(3, 2), + input->Slice(5, 1)}; + ASSERT_OK_AND_ASSIGN(std::vector> messages, + WriteArrays(table_path, {}, {}, batches)); + ASSERT_EQ(messages.size(), 1); + auto message = std::dynamic_pointer_cast(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, ".vector.vortex")); + ASSERT_EQ(files[i]->row_count, files[i + 1]->row_count); + ASSERT_EQ(files[i]->write_cols, std::vector({"id", "tag"})); + ASSERT_EQ(files[i + 1]->write_cols, std::vector({"embedding"})); + } + ASSERT_OK(Commit(table_path, messages)); + auto expected = checked_pointer_cast(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::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::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> splits, PlanSplits(table_path)); + std::map>> row_groups; + for (const auto& split : splits) { + auto data_split = std::dynamic_pointer_cast(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 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 invalid_context, invalid_builder.Finish()); + ASSERT_NOK_WITH_MSG(FileStoreWrite::Create(std::move(invalid_context)).status(), + "requires row-tracking.enabled and data-evolution.enabled"); +} +#endif + TEST_P(DataEvolutionTableTest, TestVectorReadWrite) { if (FileFormat() != "parquet" && FileFormat() != "lance") { return; diff --git a/test/inte/write_and_read_inte_test.cpp b/test/inte/write_and_read_inte_test.cpp index abff6ec75..441c6d81f 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 options = { + {Options::FILE_FORMAT, "vortex"}, {Options::BUCKET, "-1"}, {Options::FILE_SYSTEM, "local"}}; + ASSERT_OK_AND_ASSIGN( + std::unique_ptr helper, + TestHelper::Create(dir->Str(), arrow::schema(fields), /*partition_keys=*/{}, + /*primary_keys=*/{}, options, /*is_streaming_mode=*/false)); + ASSERT_OK_AND_ASSIGN(std::unique_ptr 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> 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> 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 options = { + {Options::FILE_FORMAT, "vortex"}, {Options::BUCKET, "1"}, {Options::FILE_SYSTEM, "local"}}; + ASSERT_OK_AND_ASSIGN( + std::unique_ptr helper, + TestHelper::Create(dir->Str(), arrow::schema(fields), /*partition_keys=*/{}, + /*primary_keys=*/{"pk"}, options, /*is_streaming_mode=*/true)); + const std::vector 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 batch, + TestHelper::MakeRecordBatch(arrow::struct_(fields), data_json, /*partition_map=*/{}, + /*bucket=*/0, {})); + ASSERT_OK_AND_ASSIGN(std::vector> 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> 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) { From 71c4587117d0769ca697aeb383dd503481e119b9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=B0=8F=E6=98=8E=E5=90=8C=E5=AD=A6?= <72908278+ChaomingZhangCN@users.noreply.github.com> Date: Fri, 16 Oct 2026 11:14:56 +0800 Subject: [PATCH 2/2] fix(vortex): respect null ancestors and test Java vector compatibility --- .../format/vortex/vortex_file_format_test.cpp | 46 ++++++ .../format/vortex/vortex_format_writer.cpp | 43 ++++- test/inte/data_evolution_table_test.cpp | 72 +++++++++ .../GenerateDedicatedVectorFixture.java | 147 ++++++++++++++++++ .../dedicated_vector_java_compat/README.md | 112 +++++++++++++ ...0850-170e-4085-bc42-a69c6d6fffdc-0.parquet | Bin 0 -> 475 bytes ...70e-4085-bc42-a69c6d6fffdc-1.vector.vortex | Bin 0 -> 2436 bytes ...a451-e0e6-4a75-8a36-2ea3d2387dbe-0.parquet | Bin 0 -> 476 bytes ...0e6-4a75-8a36-2ea3d2387dbe-1.vector.vortex | Bin 0 -> 2436 bytes ...est-16ae6a76-0eec-45b3-a1e2-f18249238804-0 | Bin 0 -> 2254 bytes ...est-736f9660-1f51-41e2-b0bf-751a67514c1d-0 | Bin 0 -> 2253 bytes ...ist-d1d44abf-1ad2-4f22-bd0d-bc8b0d1ab444-0 | Bin 0 -> 1261 bytes ...ist-d1d44abf-1ad2-4f22-bd0d-bc8b0d1ab444-1 | Bin 0 -> 1263 bytes ...ist-d7544098-f6d3-4754-8d1c-cc9e672f1970-0 | Bin 0 -> 1158 bytes ...ist-d7544098-f6d3-4754-8d1c-cc9e672f1970-1 | Bin 0 -> 1261 bytes .../dedicated_vector_java_compat/pom.xml | 56 +++++++ .../schema/schema-0 | 33 ++++ .../snapshot/LATEST | 1 + .../snapshot/snapshot-1 | 18 +++ .../snapshot/snapshot-2 | 18 +++ 20 files changed, 544 insertions(+), 2 deletions(-) create mode 100644 test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/GenerateDedicatedVectorFixture.java create mode 100644 test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/README.md create mode 100644 test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/bucket-0/data-02c40850-170e-4085-bc42-a69c6d6fffdc-0.parquet create mode 100644 test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/bucket-0/data-02c40850-170e-4085-bc42-a69c6d6fffdc-1.vector.vortex create mode 100644 test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/bucket-0/data-bd5ba451-e0e6-4a75-8a36-2ea3d2387dbe-0.parquet create mode 100644 test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/bucket-0/data-bd5ba451-e0e6-4a75-8a36-2ea3d2387dbe-1.vector.vortex create mode 100644 test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-16ae6a76-0eec-45b3-a1e2-f18249238804-0 create mode 100644 test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-736f9660-1f51-41e2-b0bf-751a67514c1d-0 create mode 100644 test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-list-d1d44abf-1ad2-4f22-bd0d-bc8b0d1ab444-0 create mode 100644 test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-list-d1d44abf-1ad2-4f22-bd0d-bc8b0d1ab444-1 create mode 100644 test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-list-d7544098-f6d3-4754-8d1c-cc9e672f1970-0 create mode 100644 test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-list-d7544098-f6d3-4754-8d1c-cc9e672f1970-1 create mode 100644 test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/pom.xml create mode 100644 test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/schema/schema-0 create mode 100644 test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/snapshot/LATEST create mode 100644 test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/snapshot/snapshot-1 create mode 100644 test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/snapshot/snapshot-2 diff --git a/src/paimon/format/vortex/vortex_file_format_test.cpp b/src/paimon/format/vortex/vortex_file_format_test.cpp index dce351f69..4ac9c2662 100644 --- a/src/paimon/format/vortex/vortex_file_format_test.cpp +++ b/src/paimon/format/vortex/vortex_file_format_test.cpp @@ -497,6 +497,52 @@ TEST_F(VortexFileFormatTest, RejectNullVectorElements) { "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 parent; + if (list_parent) { + auto offsets = + arrow::ipc::internal::json::ArrayFromJSON(arrow::int32(), "[0, 1, 2, 3, 4]") + .ValueOrDie(); + parent = std::make_shared(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>. TEST_F(VortexFileFormatTest, WriteThenReadListOfStruct) { diff --git a/src/paimon/format/vortex/vortex_format_writer.cpp b/src/paimon/format/vortex/vortex_format_writer.cpp index e265c593d..0dfffcd17 100644 --- a/src/paimon/format/vortex/vortex_format_writer.cpp +++ b/src/paimon/format/vortex/vortex_format_writer.cpp @@ -21,6 +21,7 @@ #include +#include "arrow/array/array_nested.h" #include "arrow/array/data.h" #include "arrow/c/bridge.h" #include "arrow/compute/api.h" @@ -32,6 +33,7 @@ #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" @@ -39,6 +41,43 @@ namespace paimon::vortex { namespace { +Status ValidateVisibleVectorElements(const std::shared_ptr& 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(*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(*array); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr values, + list.Flatten(pool)); + return ValidateVisibleVectorElements(values, pool); + } + case arrow::Type::FIXED_SIZE_LIST: { + const auto& list = checked_cast(*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 values, + list.Flatten(pool)); + return ValidateVisibleVectorElements(values, pool); + } + default: + return Status::OK(); + } +} + Result> NormalizeVectorNullElements( const std::shared_ptr& data, arrow::compute::ExecContext* context) { std::shared_ptr normalized = data; @@ -47,8 +86,7 @@ Result> NormalizeVectorNullElements( 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 may own null element slots. Vortex imports these masked slots too. - PAIMON_RETURN_NOT_OK(VectorUtils::ValidateVectorElements(*arrow::MakeArray(data))); + // Null VECTORs or ancestors may own null slots. Vortex imports these masked slots too. PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr zero, arrow::MakeScalar(child->type, 0)); PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( @@ -137,6 +175,7 @@ Status VortexFormatWriter::AddBatch(::ArrowArray* batch) { arrow::ImportArray(batch, arrow::struct_(schema_->fields()))); PAIMON_ASSIGN_OR_RAISE(std::shared_ptr 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 normalized_data, NormalizeVectorNullElements(normalized->data(), &context)); diff --git a/test/inte/data_evolution_table_test.cpp b/test/inte/data_evolution_table_test.cpp index 5939e3d39..b1d0372b9 100644 --- a/test/inte/data_evolution_table_test.cpp +++ b/test/inte/data_evolution_table_test.cpp @@ -1798,6 +1798,78 @@ 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::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(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(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> splits, PlanSplits(table_path)); + std::map>> row_groups; + for (const auto& split : splits) { + auto data_split = std::dynamic_pointer_cast(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({"embedding"})); + } else { + ASSERT_TRUE(StringUtils::EndsWith(file->file_name, ".parquet")); + ASSERT_EQ(file->write_cols, std::vector({"id", "tag"})); + } + } + ASSERT_EQ(vector_file_count, 1); + } +} + TEST_P(DataEvolutionTableTest, TestDedicatedVectorStore) { if (FileFormat() != "parquet") { GTEST_SKIP() << "Dedicated VECTOR storage requires parquet main files"; diff --git a/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/GenerateDedicatedVectorFixture.java b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/GenerateDedicatedVectorFixture.java new file mode 100644 index 000000000..0e801171d --- /dev/null +++ b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/GenerateDedicatedVectorFixture.java @@ -0,0 +1,147 @@ +/* + * 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. + */ + +import org.apache.paimon.catalog.Catalog; +import org.apache.paimon.catalog.CatalogContext; +import org.apache.paimon.catalog.CatalogFactory; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.BinaryVector; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.manifest.ManifestEntry; +import org.apache.paimon.options.Options; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.Table; +import org.apache.paimon.table.sink.BatchTableCommit; +import org.apache.paimon.table.sink.BatchTableWrite; +import org.apache.paimon.table.sink.BatchWriteBuilder; +import org.apache.paimon.table.source.ReadBuilder; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.utils.CloseableIterator; + +import java.util.Arrays; +import java.util.Map; +import java.util.TreeMap; + +/** Generates the Java-written table consumed by the C++ interoperability test. */ +public class GenerateDedicatedVectorFixture { + public static void main(String[] args) throws Exception { + if (args.length != 1) { + throw new IllegalArgumentException("Expected an empty warehouse directory argument"); + } + Options options = new Options(); + options.set("warehouse", args[0]); + Identifier identifier = + Identifier.create("dedicated_vector_java_compat", "dedicated_vector_java_compat"); + try (Catalog catalog = CatalogFactory.createCatalog(CatalogContext.create(options))) { + catalog.createDatabase(identifier.getDatabaseName(), false); + Schema schema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("embedding", DataTypes.VECTOR(3, DataTypes.FLOAT())) + .column("tag", DataTypes.STRING()) + .option("file.format", "parquet") + .option("vector.file.format", "vortex") + .option("bucket", "-1") + .option("row-tracking.enabled", "true") + .option("data-evolution.enabled", "true") + .option("write-only", "true") + .build(); + catalog.createTable(identifier, schema, false); + Table table = catalog.getTable(identifier); + append( + table, + row(1, new float[] {1, 2, 3}, "a"), + row(2, null, "b"), + row(3, new float[] {-1, 0, 1}, "c")); + append( + table, + row(4, null, "d"), + row(5, new float[] {7, 8, 9}, "e"), + row(6, new float[] {10, 11, 12}, "f")); + + ReadBuilder read = table.newReadBuilder(); + Map actual = new TreeMap<>(); + try (CloseableIterator rows = + read.newRead().createReader(read.newScan().plan()).toCloseableIterator()) { + while (rows.hasNext()) { + InternalRow row = rows.next(); + String vector = + row.isNullAt(1) + ? "null" + : Arrays.toString(row.getVector(1).toFloatArray()); + actual.put(row.getInt(0), vector + ":" + row.getString(2)); + } + } + Map expected = new TreeMap<>(); + expected.put(1, "[1.0, 2.0, 3.0]:a"); + expected.put(2, "null:b"); + expected.put(3, "[-1.0, 0.0, 1.0]:c"); + expected.put(4, "null:d"); + expected.put(5, "[7.0, 8.0, 9.0]:e"); + expected.put(6, "[10.0, 11.0, 12.0]:f"); + if (!actual.equals(expected)) { + throw new AssertionError("Unexpected Java readback: " + actual); + } + System.out.println("Java readback verified: " + actual); + Map ranges = new TreeMap<>(); + for (ManifestEntry entry : ((FileStoreTable) table).store().newScan().plan().files()) { + DataFileMeta file = entry.file(); + if (file.rowCount() != 3) { + throw new AssertionError("Unexpected file row count: " + file); + } + ranges.merge(file.nonNullFirstRowId(), 1, Integer::sum); + System.out.println( + file.fileName() + + " firstRowId=" + + file.firstRowId() + + " rowCount=" + + file.rowCount() + + " writeCols=" + + file.writeCols()); + } + Map expectedRanges = new TreeMap<>(); + expectedRanges.put(0L, 2); + expectedRanges.put(3L, 2); + if (!ranges.equals(expectedRanges)) { + throw new AssertionError("Expected one main/vector pair per commit: " + ranges); + } + } + } + + private static GenericRow row(int id, float[] vector, String tag) { + return GenericRow.of( + id, + vector == null ? null : BinaryVector.fromPrimitiveArray(vector), + BinaryString.fromString(tag)); + } + + private static void append(Table table, InternalRow... rows) throws Exception { + BatchWriteBuilder builder = table.newBatchWriteBuilder(); + try (BatchTableWrite write = builder.newWrite(); + BatchTableCommit commit = builder.newCommit()) { + for (InternalRow row : rows) { + write.write(row); + } + commit.commit(write.prepareCommit()); + } + } +} 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 000000000..aa5d550c4 --- /dev/null +++ b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/README.md @@ -0,0 +1,112 @@ + + +# Dedicated VECTOR Java Compatibility Fixture + +This table was generated on 2026-10-16 by the Paimon Java batch write API, +then read back and checked by Java before copying the table files here. +The C++ `DataEvolutionTableTest.TestReadJavaDedicatedVectorStore` test reads +these files without rewriting them. Metadata was generated by Java together +with the data files. + +Writer provenance: + +- Apache Paimon Java `2.2-SNAPSHOT`, source commit + [`7d6d146bbaf33a349dc83c3caf71334bd551e3d5`](https://github.com/apache/paimon/commit/7d6d146bbaf33a349dc83c3caf71334bd551e3d5). +- Vortex native library from `dev.vortex:vortex-jni:0.73.0`, as selected by + that Java source revision; Arrow Java `15.0.0`. +- Amazon Corretto JDK `17.0.20.1` (`17.0.20.12.1`), Maven `3.9.16`, + macOS `15.5`, `aarch64`. +- `GenerateDedicatedVectorFixture.java` is the generator used for this table. + `pom.xml` supplies its runtime classpath after building the pinned revision. + +Schema (all three columns are nullable): + +```text +id INT +embedding VECTOR +tag STRING +``` + +Options: + +```text +file.format = parquet +vector.file.format = vortex +bucket = -1 +row-tracking.enabled = true +data-evolution.enabled = true +write-only = true +``` + +`write-only` prevents background compaction so that the two append commits +remain distinct. Every non-null vector has three non-null FLOAT elements. + +| Snapshot | id | embedding | tag | +|---|---|---|---| +| 1 | 1 | [1, 2, 3] | a | +| 1 | 2 | NULL | b | +| 1 | 3 | [-1, 0, 1] | c | +| 2 | 4 | NULL | d | +| 2 | 5 | [7, 8, 9] | e | +| 2 | 6 | [10, 11, 12] | f | + +The table has two Parquet main files with `writeCols=[id, tag]` and two +`.vector.vortex` files with `writeCols=[embedding]`. The first main/vector pair +has `firstRowId=0`; the second pair has `firstRowId=3`. Each file has three +rows. The 14 Java-generated table files total 17,178 bytes, excluding this +README and the generator sources. + +## Reproduce + +Run from the Paimon C++ repository root with JDK 17 and Maven available. +This creates a new temporary warehouse and leaves the checked-in fixture +untouched. + +```bash +fixture_dir="$PWD/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat" +generation_dir="$(mktemp -d)" +git clone https://github.com/apache/paimon.git "$generation_dir/paimon" +git -C "$generation_dir/paimon" checkout 7d6d146bbaf33a349dc83c3caf71334bd551e3d5 + +mvn -B -ntp -f "$generation_dir/paimon/pom.xml" \ + -pl paimon-vortex/paimon-vortex-format -am \ + -DskipTests -Dcheckstyle.skip -Dspotless.check.skip \ + -Drat.skip=true -Dlicense.skip=true install + +mvn -B -ntp -f "$fixture_dir/pom.xml" \ + org.apache.maven.plugins:maven-dependency-plugin:3.6.1:build-classpath \ + -Dmdep.outputFile="$generation_dir/classpath.txt" + +java --add-opens=java.base/java.nio=ALL-UNNAMED \ + --class-path "$(cat "$generation_dir/classpath.txt")" \ + "$fixture_dir/GenerateDedicatedVectorFixture.java" \ + "$generation_dir/warehouse" +``` + +The Arrow Java memory implementation requires the `--add-opens` argument. +The generator creates the database and table, appends each three-row batch +in a separate commit, and verifies all six rows, NULL vectors, and paired +row-range metadata with Java. A successful run prints `Java readback verified` +and the four file metadata records, and exits with code zero. + +The generated table is under +`$generation_dir/warehouse/dedicated_vector_java_compat.db/dedicated_vector_java_compat/`. +UUIDs, timestamps, and commit users change when regenerating; row values and +row-range relationships remain the same. 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 0000000000000000000000000000000000000000..2a1a3aac545f097c748474004498bec848242d28 GIT binary patch literal 475 zcmZWmy-or_5S~5oyaEdoW^-$ZcnN?bJC@ItgU#D3hOct#Z7B_G} zFCZ+QVxllyS7^7hHTL4+D7l>Y*Wu8^EM-fYsl~xL$>r#pnR-_xXl5;o_*@s3sFuMZ zZc_(CC?w{%&U<`NHr!PktDZPimQAf9Dos<8I=CeBtYMyiFsy;n%B{6jsioKjPs*w5 zK!rsn3;s~4Yv6Ah?(hc!6ei%7&PgH4IHE@yE+@){*0YGI) As{jB1 literal 0 HcmV?d00001 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 0000000000000000000000000000000000000000..b52459a44afcfc9c5237a03b4e0ea8f1b256b2fc GIT binary patch literal 2436 zcmaJ@JBSle6uoi7e&QMC%z~K4vyA67=0|}tFad~clLb9;W~8Kts?5uzwyVjURJCrWtbv@z zyav!$S^sHL+uzVSkz<3wKwux?Jph(~{eZ|G@nFfUrnDF8=AfmNN{8^|I{MEGbMx@v zSVIna#GJWu?&OJ%yGPzZaU56#T-?n*nLXp0I$#2!KMKb=J*&#TLLDY$1>7azJmCEA z|IfZyW7dsl;e>g6lUwh}<>A(Hd}EcK9XFpV^X0iZTl)W?>#kBslB&G`kEr>)_t=mFD{P*u$NRB-m})+wn#2`N#}GcwxY z`U&HMy+kjRnG$_7hZ#V%8!8CG_%i0^EE$)(beVxb_c;H2qtFL+X06GF7G$XBIl7*~ z@rigMbquWcBEa}xQ3Kl07loslSG5n5Z&}|fu5p4cL2RklBu;LX_j=G8Qwozj&X;apQU(} zzn0rR;#!fPiHG;^2=WTp0M>yO;0$mWm<7J0XzzdxU>#TjjsqU>6Px)6yaXNqH-G>* z11tb#;0t2B0ycnk;1W;=JfIBR`)&4=%dMYgfjqPN++O?P<-X5ed2cy>^Xy)GeiHgI z3h~WyyT-Ui823VT4Z6yoLJeUH9^AnqH>~f;Ex1)bNfqoaT5jhkW~#1sRoqn4FWM#D zSj7R>-mx5e*6w393EN>7_UKlgTb}pK@y!nQZ}qvF#oZ{9ufi zm}a^v2P~KKM^26v^{o{8vsTYwEpFxNYerWZ3 zM?M494BLLR+_;{pWz9uQiTV&H;x3_A|e0^I)b&zY`NLQ1}HF-^w@SjOj!jk1skHi3&WA a+dYN$XpW!T#NP#fpUFc?@<1-&|F3^nPmq`Z literal 0 HcmV?d00001 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 0000000000000000000000000000000000000000..4d5d1d344c806cfa1cf74a9c4b706b9dc643451b GIT binary patch literal 476 zcmZWm%}N4M6h3pcX%@B!JQr`wLPCcr%;sv22D7P^v=Sj`7cBMbULa`gsuySz zts>|Zf;K%s%U&ab&Q%vdoAYj%Yb<~y!Sik^3?%y9Q8%;`0v;f@K{U2z~0T=+J z)(x_Yh&;uF+QnmhzPtTi`$u~GOA0kJ(ngUHn@r@=IwO}NhfMHQ*$73>P^*H?E$q_^ z2%9H3));PTbkN)whrPjQcs@~=@gT%BRf{agdYw~}%TSn_h8IPsrEQz|+!VGbIbajF zW`ZR&5;NT7ZQggRU^&KeCJydmsAWd2X-!gxfMi}6X8D(aWzbrgm6mR`5If^zCkcFT z4V5#(KUA6;_#4A5K4-WM{K?>^WARTTOE*3lhspIUp6(5j{7sxpdZS@pmZe|z@@_Qi yU3KyYUPV=ilc66%k7-D-jE^v=IEP=e*DEY$kY&CNi!n`Y^+waNc;ns3|Yn2r{Zar7#%X4+M^#4)U#n1EwdTDw*>s;q2M1V^V?=QlOq^ zWVHSDBgO}NiC!oZ1^Q+V6M$+|mG1}9HO$RfGA?)NG6R9`asK&6p%3cJT9XYe$WYI7 zbUlOP6RC*IVPL&a0F3_?HJ}ZBQ8=1;S^F^g7WKX2I_J-=H}QD?ipbaaBCK{kO6U=>&bE&#`XN#Hw*_8!;(R)KloG*AJ4Vl$tBSHL>Z z1$^KFFbfocFNm=TYyhjkRbURN07c-zZ?mUNZk;6c<(buI_Sy-qbv*XU2g~uBXZPCn z;=qegh;NqLJH%Cks2!-A(3Sq=Y6x5K;Ev?Eetpkw!!3JpqF}dRxjloJiMr8JQB6rF zZx^>B75P|u&vNWpyN~5KXas4{rdxS#dEPU}*P7VB)n{rJwZc%o=3_OyTf zGdYGqlIp4)wp_*^IXRWr*AwV3T0OtbHm`cHqWr}q=%__)DzROY$z84Bd~WA``T;ul zq17K4`1DmRXn5gb^;W7DQET1I@9etF+1quv)0X4ivU&Mls?wmLoVk>6@7Z^mI@Fp` zlCpV@ST5tcw$zGr0ng^`m{+yg3}wM`nfhsc+v+oS(hHYk9*5kuT&6zYL>^l``@Zfa zY3M~NIUa{>h}YY4^=1t2v*oz2ft|0y)PrXDJGF0UO*KfDy&5-(D{a5iB0q3$!(CWJtVe-Q^_LIOr(*}i4Xx(>Q-pkyg!E3BmZNmq;rOW;B< z=+(p!5)OJGE*^}cUX+EKQ9~jmB8ib04OcbFfuMeEx3;W|VdBNxw(ohL?|q-|dEU8t z`tu48GgpO(IpzIKSbszaAazKX2^qe}qKseqF(UMt8TdKKnxhOC ziiFKFSsxo_1x940@?MVR12LqfYbeb`djaXjTm(E!-Nl7PBwnWh92jBEe%gkT6h>_! z=ou8**cOUbkHrC~>?D+nLUCF_xwFG;7!K6a9NN%y3cbz9(g{tVPK^wSi2@hy+tRd* za${F?C_a*bAzdA`#pNM%vl%c>jG!Ihvf4Oe)W?Z-=7K;5+D)O9CeN+}pe0_la2qY8 z?z|}O0UB|Z26Lkb4~{ExXk=YTc!1{-=%&+-Pq0XaFs?^)3L~#PGOsSUs|58XV=LYQ_KvWzI}%>L>t|-)aR` zd0-lDOwg%Hqp~Q$->DO2=eRtiMcX=Rxuoz;#ky?36ylofX$DoZZ1V5K4&M`y1YFZ? zC}lU!mRg`YG1xHDP01?je3kAkx5OPJ1|R9)52o>|0r?B*lvo(XhFi-QFPHy+{gD9c zKX4t|F&toofT<~LiX?(1-Lp?RrVjq7WD3U$PCl;RG})JmOvuaH(<7&_;&Xd%PTg2t zJNj_u(C9IH`JR(ywGqvL^p+BJjHG z!o<&Rv-eXnh?WErwx(y5if`PF&bbb`n$Gqpv7jc@u_tdp?JI|3?MeUu literal 0 HcmV?d00001 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 0000000000000000000000000000000000000000..f7a3604ddb74079517059647d618a286a5e9fa36 GIT binary patch literal 2253 zcmeZI%3@>@ODrqO*DFrWNXij}OQt6?U^hq(p?d;0qUC82g@ z=9MVb>L3(&d-}QHmJM(W3h@j-~-QA$ZoODxSP zQL+N*tc|TjPK;nj!2JU;5^gm1Tmnf6@ep6awGRED0L5IeYhb9WpEEEuhx$6X1|iZI zW-0_GG2&GRJA1hLI>vhnLR!|}ip(=`GfG;mWO86TfZAZ3n!Xpl3a z)PZXUg)lzNNO=R83w;p54l@BBX1IbMpXtE(bPe(Y7SjQaAs(21gBGCpOmOoI3Jw8> zGE%yNxs7lThX;8ABZ>MRLzoNCAS47GBwJv38(3<^){21Z-{kz1)MTM5U~P`4;}_d#LsG1BGAzowJ^OHtlsmKq>=6E&mZyiRW{Ca~>?J z^o@LQS>m{^ua)a1na4X=wbt$Z^~mH*7^5J=hSZErr?c8!W*0M;=$n0Rw0>Z;z^|eE W-~ol=1Oo#Ji8TirWE7atJp}+!?(U@k literal 0 HcmV?d00001 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 0000000000000000000000000000000000000000..4896c71a87f905843e52e573297a97a81d50ab7b GIT binary patch literal 1261 zcmbVM%Wl&^6rBct052id!={TkBxwjCJ2xIw3qMe7hl}^=iV7@{C10bK;E&i5!r{-C#QFgVLYEv z*oTbTA#-3I2E?y?M>Aq)KSMTcko5hKHfF^2!=SM!Hl+b&B&HdC;!@9vVE^EFRSA#K zP?t0{U0iCpK@5|64tOrf2&;mEa!3?2H<(sYy7an$^1>2EH5Dv~A`)>{Q_t&Ni4>8H zj2mEI3PVFOQVUo&1S79{O?VhriC4sCxJVmQOvixgm%f(F5jfo<~vWuJ3ML8Ha}_g+MQ?Z gonEim;hP^Hao}P4)Bvb)0Netfl@DL0V7n~y8&O)7?EnA( literal 0 HcmV?d00001 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 0000000000000000000000000000000000000000..cc700356e653a3ea0c2e789d207277bd1aecd21c GIT binary patch literal 1263 zcmbVM%Wj)M6rE73i|!g#QM;XHk%e(hn>dQ4EQpyZs5~kIZq&#fjbR`}Jgk94awDXC zM^=@(?KU6KU+KP|P^sYBhC_u98c{}4TF_me22KL|r&r5Lc!Y+A ztZVt=Ld%a*n6*p5TUkL^6IE0~l9c(8*F+ie#|kQlYZ%Q^u_#GM!bMG^tam3;LJBe; zfPF6xP1(#XV8ay6vg$qIVcaC%5sxZscqA%Tb-R;UN|V3btK!K6P@1JEc%WNaWtXQg zx(EW8=e>xZFh*v1G=56MqEm}?IL~ODjoZ+nF1ZL&*e{sbIIA|x7#t&2#Ip3Ca0^RE z*5Dm7SD=+j(fo!)Wf%&hYjbm51&xrhp6t5RTy>~_s7A7K`zO zxOMJu-!^t`U;P5dAqTcwtxq9|e3vFEzyGG^9`t%$UU2sWepjHc`AK);^80%N>E*Xw gTX6X9&C3>eo?jdQ++*$u0FS`I(e-x^?9_#R1MmlrrvLx| literal 0 HcmV?d00001 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 0000000000000000000000000000000000000000..122e2d8cf9c7c09c428259d8d6caacaa46f36ca9 GIT binary patch literal 1158 zcmbW1PfmkC6vj1i=|*Ebz^t3bHM_BlO5~3zL))|=;}oVXhC(w;O=`p5Gk6V8;1Rrp zH*n)hs1y*YZMp;I``$P2_hvAL4~IwR^pS_n$eFMyeK}bIv6wL+0nVI|yPyFAI<37& zGwP%tLq2ZOCnD0hF8Z6h9NC&k|#n6+*EXxZ7h#Q6A zN>LHfp@s@cB)Auh>nc;}msEaO(dd>&pe!Sqq&3Z=-bSU2RD^fX-oZ9jaMA)XF~mjH zE$|4b8#mz56rFUSYL&M;TnHBZ>RyAr9aOrds-$IDdTEyf5X^laB)l{HlykaB(0GDP z(@ACPV8zIe?FMk!h|YZhq|_PPad}u~>jr6%DDppW)0Qr+)-A%hC{Ia;UyCRmhP2VW zy-BW$I!N6wc3*3f+BUjnBzd<#OshmI?e1;H28L=zt2mZ<_%_t2hug4nQ`Y9cP77Gy o%-Rg?_-yqza6*?kr%#cfflIl2aQ*r5b^m_V%^i|u>w_wK#l?@k@R+~RK2(=cd6)-m(wo15n_p3fNU z!;o1)Xu~@6saN@qX4J}l2H~_p)AwW6m{G?I{Klf#l=&>AG0W%^hq-nH`$y-iN@RqF znxrb};!?};W0=%)z+(v`q6jj|AyFJU{AQI$UIACT1PlY(h`m81^$}VkX6AX@h+v3q%tC6K-bd zkuf+x`Wm!W$(mo0C=5epbYpI+i_r+-&15&Frm92rX)%(u+dr9>6D`@@n5?V1fR}0I zW4RW;Dr$774~5c6S&{!WEpT-*D?K!moyBcn1vay8okcPAZ5rC(yZO%GUVnY{{s+GC z@yoAgwGUgnpYHtx`(_Pr9QVSbzB6G_%(uFfb?JjH-(<|QY=h78#kVQgEzA4{$byr< literal 0 HcmV?d00001 diff --git a/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/pom.xml b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/pom.xml new file mode 100644 index 000000000..3e6407744 --- /dev/null +++ b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/pom.xml @@ -0,0 +1,56 @@ + + + + 4.0.0 + org.apache.paimon.fixture + dedicated-vector-java-compat + 1.0 + + 2.2-SNAPSHOT + + + + org.apache.paimon + paimon-core + ${paimon.version} + + + org.apache.paimon + paimon-format + ${paimon.version} + + + org.apache.paimon + paimon-vortex-format + ${paimon.version} + + + org.apache.hadoop + hadoop-mapreduce-client-core + 2.8.5 + + + org.slf4j + slf4j-simple + 1.7.32 + + + 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 000000000..4a7d5c836 --- /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 000000000..d8263ee98 --- /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 000000000..13f943658 --- /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 000000000..f3767d9fe --- /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