fix(format): address second-round review (flush serialization, pool, validation, CI deps)

- Serialize host OutputStream access between the Vortex writer task and a
  caller-thread Flush(): VortexOutputContext gains a stream mutex and a
  FlushStream() entry, and VortexFormatWriter::Flush() plus the write/flush
  callbacks all take it, so Write and Flush can no longer interleave on a stream
  with no concurrent-access contract. The remaining internally buffered data is
  drained when the sink closes in Finish().
- Thread the caller's Arrow memory pool into VortexFormatWriter (Create now takes
  it and the builder passes GetArrowPool(pool_)), so sliced-batch rebasing
  allocates from the Paimon pool instead of Arrow's default pool.
- Validate Vortex-incompatible schema types wherever Vortex can be selected
  (default file format, file.format-per-level, changelog-file.format), mirroring
  the Lance validation, instead of only the default format.
- setup_flatc.sh: install libclang before the flatc early-exit, so an environment
  that already has a matching flatc but lacks libclang no longer silently skips
  the bindgen build dependency.
- Tests: FlushBetweenBatchesThenRead (flush interleaved with background writes)
  and SchemaValidationTest.TestVortexDataTypes (per-level/changelog validation).

Author: zhouyc-ali

ci/scripts/setup_flatc.sh

@@ -36,6 +36,18 @@ set -eux
 FLATBUFFERS_VERSION=${FLATBUFFERS_VERSION:-25.12.19}
 FLATC_INSTALL_DIR=${FLATC_INSTALL_DIR:-"${HOME}/.local/bin"}
 
+# Vortex's native dependencies invoke bindgen in their build scripts, which loads libclang at
+# build time. Images that only carry GCC (e.g. the gcc8 test container) may lack it.
+if ! ldconfig -p 2>/dev/null | grep -q 'libclang'; then
+    if [[ "$(id -u)" -ne 0 ]] && command -v sudo >/dev/null 2>&1; then
+        apt_prefix="sudo"
+    else
+        apt_prefix=""
+    fi
+    ${apt_prefix} apt-get update -y
+    ${apt_prefix} apt-get install -y libclang-dev
+fi
+
 # Skip when a matching flatc is already available.
 if command -v flatc >/dev/null 2>&1 &&
     [[ "$(flatc --version 2>/dev/null | awk '{print $NF}')" == "${FLATBUFFERS_VERSION}" ]]; then
@@ -61,16 +73,4 @@ export PATH="${FLATC_INSTALL_DIR}:${PATH}"
 echo "${FLATC_INSTALL_DIR}" >>"${GITHUB_PATH:-/dev/null}" || true
 echo "FLATC=${FLATC_INSTALL_DIR}/flatc" >>"${GITHUB_ENV:-/dev/null}" || true
 
-# Vortex's native dependencies invoke bindgen in their build scripts, which loads libclang at
-# build time. Images that only carry GCC (e.g. the gcc8 test container) may lack it.
-if ! ldconfig -p 2>/dev/null | grep -q 'libclang'; then
-    if [[ "$(id -u)" -ne 0 ]] && command -v sudo >/dev/null 2>&1; then
-        apt_prefix="sudo"
-    else
-        apt_prefix=""
-    fi
-    ${apt_prefix} apt-get update -y
-    ${apt_prefix} apt-get install -y libclang-dev
-fi
-
 "${FLATC_INSTALL_DIR}/flatc" --version

src/paimon/core/schema/schema_validation.cpp

@@ -935,21 +935,35 @@ Status SchemaValidation::ValidateVortexDataField(const std::shared_ptr<arrow::Fi
 
 Status SchemaValidation::ValidateVortexDataFields(const TableSchema& schema,
                                                   const CoreOptions& options) {
-    if (StringUtils::ToLowerCase(options.GetFileFormat()->Identifier()) != "vortex") {
-        return Status::OK();
-    }
-
     const std::vector<std::string> inline_blob_fields = options.GetBlobInlineFields();
     const std::set<std::string> inline_blob_field_set(inline_blob_fields.begin(),
                                                       inline_blob_fields.end());
-    // Mirror the Mosaic path: only validate fields stored in the normal data file. A non-inline
-    // BLOB lives in a separate blob file, so it is skipped here; an inline BLOB is rejected.
-    for (const DataField& field : schema.Fields()) {
-        if (BlobUtils::IsBlobField(field.ArrowField()) &&
-            inline_blob_field_set.count(field.Name()) == 0) {
-            continue;
+    // Mirror the Lance path: Vortex can be selected not only as the default file format but also
+    // per level or for the changelog, so validate the schema against every place it can be chosen.
+    auto validate_format = [&](const std::string&, const std::string& file_format) -> Status {
+        if (!StringUtils::EqualsIgnoreCase(file_format, "vortex")) {
+            return Status::OK();
+        }
+        // Mirror the Mosaic path: only validate fields stored in the normal data file. A non-inline
+        // BLOB lives in a separate blob file, so it is skipped here; an inline BLOB is rejected.
+        for (const DataField& field : schema.Fields()) {
+            if (BlobUtils::IsBlobField(field.ArrowField()) &&
+                inline_blob_field_set.count(field.Name()) == 0) {
+                continue;
+            }
+            PAIMON_RETURN_NOT_OK(ValidateVortexDataField(field.ArrowField()));
         }
-        PAIMON_RETURN_NOT_OK(ValidateVortexDataField(field.ArrowField()));
+        return Status::OK();
+    };
+
+    PAIMON_RETURN_NOT_OK(
+        validate_format(Options::FILE_FORMAT, options.GetFileFormat()->Identifier()));
+    PAIMON_RETURN_NOT_OK(
+        ValidatePerLevelOption(options.ToMap(), Options::FILE_FORMAT_PER_LEVEL, validate_format));
+    std::shared_ptr<FileFormat> changelog_format = options.GetChangelogFileFormat();
+    if (changelog_format) {
+        PAIMON_RETURN_NOT_OK(
+            validate_format(Options::CHANGELOG_FILE_FORMAT, changelog_format->Identifier()));
     }
     return Status::OK();
 }

src/paimon/core/schema/schema_validation_test.cpp

@@ -332,6 +332,37 @@ TEST(SchemaValidationTest, TestLanceDataTypes) {
 }
 #endif
 
+#ifdef PAIMON_ENABLE_VORTEX
+TEST(SchemaValidationTest, TestVortexDataTypes) {
+    std::shared_ptr<TableSchema> table_schema;
+    // Vortex as the default file format: unsupported types are rejected at validation time.
+    std::map<std::string, std::string> vortex_options = {{Options::BUCKET, "-1"},
+                                                         {Options::FILE_FORMAT, "vortex"}};
+    ASSERT_OK_AND_ASSIGN(
+        table_schema,
+        TableSchema::Create(
+            /*schema_id=*/0,
+            arrow::schema({arrow::field("map", arrow::map(arrow::int32(), arrow::utf8()))}),
+            /*partition_keys=*/{}, /*primary_keys=*/{}, vortex_options));
+    ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema), "type MAP");
+
+    // Vortex selected per level or for the changelog must be validated too, not only the default.
+    for (const auto& [option_key, option_value] : std::vector<std::pair<std::string, std::string>>{
+             {Options::FILE_FORMAT_PER_LEVEL, "1:vortex"},
+             {Options::CHANGELOG_FILE_FORMAT, "vortex"}}) {
+        std::map<std::string, std::string> alternate_format_options = {
+            {Options::BUCKET, "-1"}, {Options::FILE_FORMAT, "parquet"}, {option_key, option_value}};
+        ASSERT_OK_AND_ASSIGN(
+            table_schema,
+            TableSchema::Create(
+                /*schema_id=*/0,
+                arrow::schema({arrow::field("map", arrow::map(arrow::int32(), arrow::utf8()))}),
+                /*partition_keys=*/{}, /*primary_keys=*/{}, alternate_format_options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema), "type MAP");
+    }
+}
+#endif
+
 TEST(SchemaValidationTest, TestRowTracking) {
     auto f0 = arrow::field("f0", arrow::utf8());
     auto f1 = arrow::field("f1", arrow::int32());

src/paimon/format/vortex/vortex_file_format_test.cpp

@@ -387,6 +387,40 @@ TEST_F(VortexFileFormatTest, WriteThenReadSlicedStructColumn) {
     AssertReadWithBatchSizes(path, schema, expected, {1, 2, 4});
 }
 
+// Regression: Flush() between batches must be safe (it is serialized with the background writer
+// task's writes on the host OutputStream) and must not lose or reorder any data.
+TEST_F(VortexFileFormatTest, FlushBetweenBatchesThenRead) {
+    std::string path = PathUtil::JoinPath(directory_->Str(), "flush-between.vortex");
+    arrow::FieldVector fields = {arrow::field("id", arrow::int32(), false),
+                                 arrow::field("name", arrow::utf8())};
+    std::shared_ptr<arrow::Schema> schema = arrow::schema(fields);
+    std::shared_ptr<arrow::Array> written =
+        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields),
+                                                  R"([[1,"one"],[2,null],[3,"three"],[4,"four"]])")
+            .ValueOrDie();
+
+    ::ArrowSchema ffi_schema = {};
+    ASSERT_TRUE(arrow::ExportSchema(*schema, &ffi_schema).ok());
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<WriterBuilder> writer_builder,
+                         format_->CreateWriterBuilder(&ffi_schema, /*batch_size=*/2));
+    writer_builder->WithMemoryPool(pool_);
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<OutputStream> output,
+                         file_system_->Create(path, /*overwrite=*/false));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<FormatWriter> writer,
+                         writer_builder->Build(output, "zstd"));
+    for (int64_t offset = 0; offset < written->length(); offset += 2) {
+        std::shared_ptr<arrow::Array> slice = written->Slice(offset, 2);
+        ::ArrowArray ffi_array = {};
+        ASSERT_TRUE(arrow::ExportArray(*slice, &ffi_array).ok());
+        ASSERT_OK(writer->AddBatch(&ffi_array));
+        ASSERT_OK(writer->Flush());
+    }
+    ASSERT_OK(writer->Finish());
+    ASSERT_OK(output->Close());
+
+    AssertReadWithBatchSizes(path, schema, written, {1, 2, 4});
+}
+
 // The IO callback bridge must surface a paimon IO failure as that paimon error, not Vortex's
 // opaque "stream error": the callback can only hand a status code back across the FFI boundary, so
 // the reader/writer stashes the real error and prefers it. `IOHook` injects a failure at the Nth

src/paimon/format/vortex/vortex_format_writer.cpp

@@ -35,18 +35,21 @@ namespace paimon::vortex {
 VortexFormatWriter::VortexFormatWriter(std::shared_ptr<OutputStream> output,
                                        std::shared_ptr<arrow::Schema> schema, VxSessionPtr session,
                                        std::shared_ptr<VortexOutputContext> output_context,
-                                       vx_callback_sink* sink)
+                                       vx_callback_sink* sink,
+                                       std::shared_ptr<arrow::MemoryPool> arrow_pool)
     : output_(std::move(output)),
       schema_(std::move(schema)),
+      arrow_pool_(std::move(arrow_pool)),
       session_(std::move(session)),
       output_context_(std::move(output_context)),
       sink_(sink),
       metrics_(std::make_shared<MetricsImpl>()) {}
 
 Result<std::unique_ptr<VortexFormatWriter>> VortexFormatWriter::Create(
-    const std::shared_ptr<OutputStream>& output, const std::shared_ptr<arrow::Schema>& schema) {
-    if (output == nullptr || schema == nullptr) {
-        return Status::Invalid("Vortex writer requires non-null output and schema");
+    const std::shared_ptr<OutputStream>& output, const std::shared_ptr<arrow::Schema>& schema,
+    const std::shared_ptr<arrow::MemoryPool>& arrow_pool) {
+    if (output == nullptr || schema == nullptr || arrow_pool == nullptr) {
+        return Status::Invalid("Vortex writer requires non-null output, schema and arrow pool");
     }
     VxSessionPtr session(vx_session_new(), vx_session_free);
     if (session == nullptr) {
@@ -69,7 +72,7 @@ Result<std::unique_ptr<VortexFormatWriter>> VortexFormatWriter::Create(
                                    output_context->GetCallbackStatus());
     }
     return std::unique_ptr<VortexFormatWriter>(new VortexFormatWriter(
-        output, schema, std::move(session), std::move(output_context), sink));
+        output, schema, std::move(session), std::move(output_context), sink, arrow_pool));
 }
 
 VortexFormatWriter::~VortexFormatWriter() {
@@ -99,9 +102,8 @@ Status VortexFormatWriter::AddBatch(::ArrowArray* batch) {
         PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
             std::shared_ptr<arrow::Array> imported,
             arrow::ImportArray(batch, arrow::struct_(schema_->fields())));
-        PAIMON_ASSIGN_OR_RAISE(
-            std::shared_ptr<arrow::Array> normalized,
-            ArrowUtils::NormalizeArrayOffsets(imported, arrow::default_memory_pool()));
+        PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> normalized,
+                               ArrowUtils::NormalizeArrayOffsets(imported, arrow_pool_.get()));
         PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*normalized, &rebased_batch));
         vortex_batch = &rebased_batch;
     }
@@ -127,8 +129,10 @@ Status VortexFormatWriter::Flush() {
     if (finished_) {
         return Status::OK();
     }
-    // Vortex buffers internally, so this only flushes what its sink has already handed over.
-    return output_->Flush();
+    // Flush through the output context so it is serialized with the background writer task's writes
+    // (OutputStream has no concurrent Write/Flush contract). Vortex buffers internally, so this
+    // flushes the bytes handed over so far; the rest is drained when the sink closes in Finish().
+    return output_context_->FlushStream();
 }
 
 Status VortexFormatWriter::Finish() {

src/paimon/format/vortex/vortex_format_writer.h

@@ -30,6 +30,7 @@
 #include "paimon/result.h"
 
 namespace arrow {
+class MemoryPool;
 class Schema;
 }  // namespace arrow
 namespace paimon {
@@ -48,7 +49,8 @@ namespace paimon::vortex {
 class VortexFormatWriter : public FormatWriter {
  public:
     static Result<std::unique_ptr<VortexFormatWriter>> Create(
-        const std::shared_ptr<OutputStream>& output, const std::shared_ptr<arrow::Schema>& schema);
+        const std::shared_ptr<OutputStream>& output, const std::shared_ptr<arrow::Schema>& schema,
+        const std::shared_ptr<arrow::MemoryPool>& arrow_pool);
 
     ~VortexFormatWriter() override;
 
@@ -62,10 +64,13 @@ class VortexFormatWriter : public FormatWriter {
  private:
     VortexFormatWriter(std::shared_ptr<OutputStream> output, std::shared_ptr<arrow::Schema> schema,
                        VxSessionPtr session, std::shared_ptr<VortexOutputContext> output_context,
-                       vx_callback_sink* sink);
+                       vx_callback_sink* sink, std::shared_ptr<arrow::MemoryPool> arrow_pool);
 
     std::shared_ptr<OutputStream> output_;
     std::shared_ptr<arrow::Schema> schema_;
+    // The caller's Arrow pool, used for any buffer (re)allocation on the write path so it counts
+    // against Paimon's memory accounting instead of Arrow's default pool.
+    std::shared_ptr<arrow::MemoryPool> arrow_pool_;
     VxSessionPtr session_;
     std::shared_ptr<VortexOutputContext> output_context_;
     // Raw pointer on purpose: close (in Finish) and abort (in the destructor) are mutually

src/paimon/format/vortex/vortex_io_callbacks.cpp

@@ -171,8 +171,12 @@ int32_t VortexOutputContext::Write(void* ctx, const uint8_t* src, size_t length)
         return -1;
     }
     auto write_length = static_cast<int64_t>(length);
-    Result<int64_t> result =
-        context->output_->Write(reinterpret_cast<const char*>(src), write_length);
+    // The Vortex writer task is the only other accessor of `output_`; hold the stream lock so a
+    // concurrent caller-thread FlushStream cannot interleave with this write.
+    Result<int64_t> result = [&] {
+        std::lock_guard<std::mutex> stream_lock(context->stream_mutex_);
+        return context->output_->Write(reinterpret_cast<const char*>(src), write_length);
+    }();
     if (!result.ok()) {
         context->SetCallbackStatus(result.status());
         return -1;
@@ -194,12 +198,17 @@ int32_t VortexOutputContext::Flush(void* ctx) noexcept {
     if (context == nullptr) {
         return -1;
     }
-    Status status = context->output_->Flush();
+    Status status = context->FlushStream();
     if (!status.ok()) {
         context->SetCallbackStatus(status);
         return -1;
     }
     return 0;
 }
 
+Status VortexOutputContext::FlushStream() {
+    std::lock_guard<std::mutex> stream_lock(stream_mutex_);
+    return output_->Flush();
+}
+
 }  // namespace paimon::vortex

src/paimon/format/vortex/vortex_io_callbacks.h

@@ -143,11 +143,25 @@ class VortexOutputContext {
     /// @return The number of bytes written to the paimon stream so far.
     int64_t BytesWritten() const;
 
+    /// Flush the paimon stream, serialized with the background writer task's writes.
+    ///
+    /// `OutputStream` carries no thread-safety contract for concurrent `Write`/`Flush`, and
+    /// Vortex's writer task invokes the write callback from its own thread, so a caller-thread
+    /// flush must take the same lock the write callback holds. Vortex buffers internally, so this
+    /// flushes the bytes handed over so far; whatever is still buffered is drained when the sink is
+    /// closed (`Finish`).
+    ///
+    /// @return OK on success, or the flush error.
+    Status FlushStream();
+
  private:
     void SetCallbackStatus(const Status& status);
 
     std::shared_ptr<OutputStream> output_;
     mutable std::mutex mutex_;
+    // Serializes access to `output_` between the Vortex writer task (write/flush callbacks) and a
+    // caller-thread `FlushStream`; `mutex_` only guards the status/byte-count bookkeeping.
+    std::mutex stream_mutex_;
     Status callback_status_;
     int64_t bytes_written_ = 0;
 };

src/paimon/format/vortex/vortex_writer_builder.cpp

@@ -21,6 +21,7 @@
 
 #include <string>
 
+#include "paimon/common/utils/arrow/mem_utils.h"
 #include "paimon/format/vortex/vortex_format_writer.h"
 #include "paimon/status.h"
 
@@ -33,7 +34,7 @@ Result<std::unique_ptr<FormatWriter>> VortexWriterBuilder::Build(
     }
     // The paimon compression hint is not mapped to the sink; Vortex picks its own encodings.
     (void)compression;
-    return VortexFormatWriter::Create(output, schema_);
+    return VortexFormatWriter::Create(output, schema_, GetArrowPool(pool_));
 }
 
 }  // namespace paimon::vortex