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..4ac9c2662 100644 --- a/src/paimon/format/vortex/vortex_file_format_test.cpp +++ b/src/paimon/format/vortex/vortex_file_format_test.cpp @@ -450,6 +450,99 @@ TEST_F(VortexFileFormatTest, WriteThenReadFixedSizeListOfString) { AssertReadWithBatchSizes(path, schema, expected, {1, 2, 3, 4}); } +TEST_F(VortexFileFormatTest, WriteThenReadVectors) { + const std::vector> 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"); +} + +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 47d271118..0dfffcd17 100644 --- a/src/paimon/format/vortex/vortex_format_writer.cpp +++ b/src/paimon/format/vortex/vortex_format_writer.cpp @@ -21,18 +21,90 @@ #include +#include "arrow/array/array_nested.h" +#include "arrow/array/data.h" #include "arrow/c/bridge.h" +#include "arrow/compute/api.h" #include "arrow/memory_pool.h" +#include "arrow/scalar.h" #include "arrow/type.h" #include "paimon/common/metrics/metrics_impl.h" #include "paimon/common/utils/arrow/arrow_utils.h" #include "paimon/common/utils/arrow/mem_utils.h" #include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/common/utils/arrow/vector_utils.h" +#include "paimon/common/utils/checked_cast.h" #include "paimon/core/schema/schema_validation.h" #include "paimon/format/vortex/vortex_ffi_util.h" #include "paimon/fs/file_system.h" namespace paimon::vortex { +namespace { + +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; + 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 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( + 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 +175,11 @@ 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)); + 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..b1d0372b9 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,179 @@ 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"; + } + 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) { 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 000000000..2a1a3aac5 Binary files /dev/null and b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/bucket-0/data-02c40850-170e-4085-bc42-a69c6d6fffdc-0.parquet differ diff --git a/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/bucket-0/data-02c40850-170e-4085-bc42-a69c6d6fffdc-1.vector.vortex b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/bucket-0/data-02c40850-170e-4085-bc42-a69c6d6fffdc-1.vector.vortex new file mode 100644 index 000000000..b52459a44 Binary files /dev/null and b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/bucket-0/data-02c40850-170e-4085-bc42-a69c6d6fffdc-1.vector.vortex differ diff --git a/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/bucket-0/data-bd5ba451-e0e6-4a75-8a36-2ea3d2387dbe-0.parquet b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/bucket-0/data-bd5ba451-e0e6-4a75-8a36-2ea3d2387dbe-0.parquet new file mode 100644 index 000000000..4d5d1d344 Binary files /dev/null and b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/bucket-0/data-bd5ba451-e0e6-4a75-8a36-2ea3d2387dbe-0.parquet differ diff --git a/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/bucket-0/data-bd5ba451-e0e6-4a75-8a36-2ea3d2387dbe-1.vector.vortex b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/bucket-0/data-bd5ba451-e0e6-4a75-8a36-2ea3d2387dbe-1.vector.vortex new file mode 100644 index 000000000..a311aeff7 Binary files /dev/null and b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/bucket-0/data-bd5ba451-e0e6-4a75-8a36-2ea3d2387dbe-1.vector.vortex differ diff --git a/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-16ae6a76-0eec-45b3-a1e2-f18249238804-0 b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-16ae6a76-0eec-45b3-a1e2-f18249238804-0 new file mode 100644 index 000000000..48fabbdbd Binary files /dev/null and b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-16ae6a76-0eec-45b3-a1e2-f18249238804-0 differ diff --git a/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-736f9660-1f51-41e2-b0bf-751a67514c1d-0 b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-736f9660-1f51-41e2-b0bf-751a67514c1d-0 new file mode 100644 index 000000000..f7a3604dd Binary files /dev/null and b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-736f9660-1f51-41e2-b0bf-751a67514c1d-0 differ diff --git a/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-list-d1d44abf-1ad2-4f22-bd0d-bc8b0d1ab444-0 b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-list-d1d44abf-1ad2-4f22-bd0d-bc8b0d1ab444-0 new file mode 100644 index 000000000..4896c71a8 Binary files /dev/null and b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-list-d1d44abf-1ad2-4f22-bd0d-bc8b0d1ab444-0 differ diff --git a/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-list-d1d44abf-1ad2-4f22-bd0d-bc8b0d1ab444-1 b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-list-d1d44abf-1ad2-4f22-bd0d-bc8b0d1ab444-1 new file mode 100644 index 000000000..cc700356e Binary files /dev/null and b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-list-d1d44abf-1ad2-4f22-bd0d-bc8b0d1ab444-1 differ diff --git a/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-list-d7544098-f6d3-4754-8d1c-cc9e672f1970-0 b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-list-d7544098-f6d3-4754-8d1c-cc9e672f1970-0 new file mode 100644 index 000000000..122e2d8cf Binary files /dev/null and b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-list-d7544098-f6d3-4754-8d1c-cc9e672f1970-0 differ diff --git a/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-list-d7544098-f6d3-4754-8d1c-cc9e672f1970-1 b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-list-d7544098-f6d3-4754-8d1c-cc9e672f1970-1 new file mode 100644 index 000000000..d834da2e6 Binary files /dev/null and b/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/manifest/manifest-list-d7544098-f6d3-4754-8d1c-cc9e672f1970-1 differ diff --git a/test/test_data/vortex/dedicated_vector_java_compat.db/dedicated_vector_java_compat/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