fix comments for paipai

Author: lxy-9602

src/paimon/core/schema/schema_validation.cpp

@@ -197,6 +197,7 @@ Status SchemaValidation::ValidateTableSchema(const TableSchema& schema) {
 
     PAIMON_RETURN_NOT_OK(ValidateRowTracking(schema, options));
     PAIMON_RETURN_NOT_OK(ValidateBlobFields(schema, options));
+    PAIMON_RETURN_NOT_OK(ValidateMosaicDataFields(schema, options));
     PAIMON_RETURN_NOT_OK(ValidateMapStorageLayout(schema, options));
     PAIMON_RETURN_NOT_OK(ValidateVectorFields(schema, options));
     return Status::OK();
@@ -561,6 +562,70 @@ Status SchemaValidation::ValidateBlobFields(const TableSchema& schema, const Cor
     return Status::OK();
 }
 
+Status SchemaValidation::ValidateMosaicDataField(const std::shared_ptr<arrow::Field>& field) {
+    if (VariantTypeUtils::IsVariantField(field)) {
+        return Status::Invalid("Mosaic file format does not support type VARIANT");
+    }
+    if (BlobUtils::IsBlobField(field)) {
+        return Status::Invalid("Mosaic file format does not support type BLOB");
+    }
+
+    const std::shared_ptr<arrow::DataType>& type = field->type();
+    switch (type->id()) {
+        case arrow::Type::BOOL:
+        case arrow::Type::INT8:
+        case arrow::Type::INT16:
+        case arrow::Type::INT32:
+        case arrow::Type::INT64:
+        case arrow::Type::FLOAT:
+        case arrow::Type::DOUBLE:
+        case arrow::Type::DATE32:
+        case arrow::Type::STRING:
+        case arrow::Type::BINARY:
+        case arrow::Type::TIME32:
+        case arrow::Type::TIMESTAMP:
+        case arrow::Type::DECIMAL128:
+            return Status::OK();
+        case arrow::Type::LIST:
+            return ValidateMosaicDataField(type->field(0));
+        case arrow::Type::MAP: {
+            const auto& map_type = checked_cast<const arrow::MapType&>(*type);
+            PAIMON_RETURN_NOT_OK(ValidateMosaicDataField(map_type.key_field()));
+            return ValidateMosaicDataField(map_type.item_field());
+        }
+        case arrow::Type::FIXED_SIZE_LIST:
+            return Status::Invalid("Mosaic file format does not support type VECTOR");
+        case arrow::Type::STRUCT:
+            return Status::Invalid("Mosaic file format does not support type ROW");
+        default:
+            break;
+    }
+    return Status::Invalid(
+        fmt::format("Mosaic file format does not support type {}", type->ToString()));
+}
+
+Status SchemaValidation::ValidateMosaicDataFields(const TableSchema& schema,
+                                                  const CoreOptions& options) {
+    if (StringUtils::ToLowerCase(options.GetFileFormat()->Identifier()) != "mosaic") {
+        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());
+    // Match Java SchemaValidation by validating only fields stored in the normal data file. C++
+    // permits BLOB only as a top-level field; descriptor and view fields are inline, so Mosaic
+    // must reject them here.
+    for (const DataField& field : schema.Fields()) {
+        if (BlobUtils::IsBlobField(field.ArrowField()) &&
+            inline_blob_field_set.count(field.Name()) == 0) {
+            continue;
+        }
+        PAIMON_RETURN_NOT_OK(ValidateMosaicDataField(field.ArrowField()));
+    }
+    return Status::OK();
+}
+
 Status SchemaValidation::ValidateMapStorageLayout(const TableSchema& schema,
                                                   const CoreOptions& options) {
     // Extract all field names that have map.storage-layout configured from options

src/paimon/core/schema/schema_validation.h

@@ -73,6 +73,10 @@ class SchemaValidation {
 
     static Status ValidateBlobFields(const TableSchema& schema, const CoreOptions& options);
 
+    static Status ValidateMosaicDataFields(const TableSchema& schema, const CoreOptions& options);
+
+    static Status ValidateMosaicDataField(const std::shared_ptr<arrow::Field>& field);
+
     static Status ValidateMapStorageLayout(const TableSchema& schema, const CoreOptions& options);
 
     static Status ValidateVectorFields(const TableSchema& schema, const CoreOptions& options);

src/paimon/core/schema/schema_validation_test.cpp

@@ -24,6 +24,7 @@
 #include "arrow/api.h"
 #include "gtest/gtest.h"
 #include "paimon/common/data/blob_utils.h"
+#include "paimon/common/data/variant/variant_type_utils.h"
 #include "paimon/core/schema/table_schema.h"
 #include "paimon/defs.h"
 #include "paimon/testing/utils/testharness.h"
@@ -109,6 +110,74 @@ TEST(SchemaValidationTest, TestVectorType) {
                         "VECTOR fields in data-evolution tables are not implemented yet.");
 }
 
+#ifdef PAIMON_ENABLE_MOSAIC
+TEST(SchemaValidationTest, TestMosaicDataTypes) {
+    std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                  {Options::FILE_FORMAT, "mosaic"}};
+    arrow::FieldVector supported_fields = {
+        arrow::field("f0", arrow::boolean()),
+        arrow::field("f1", arrow::int8()),
+        arrow::field("f2", arrow::int16()),
+        arrow::field("f3", arrow::int32()),
+        arrow::field("f4", arrow::int64()),
+        arrow::field("f5", arrow::float32()),
+        arrow::field("f6", arrow::float64()),
+        arrow::field("f7", arrow::utf8()),
+        arrow::field("f8", arrow::binary()),
+        arrow::field("f9", arrow::date32()),
+        arrow::field("f10", arrow::timestamp(arrow::TimeUnit::NANO)),
+        arrow::field("f11", arrow::decimal128(38, 2)),
+        arrow::field("f12", arrow::list(arrow::float32())),
+        arrow::field("f13", arrow::map(arrow::int8(), arrow::int16())),
+    };
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+                         TableSchema::Create(/*schema_id=*/0, arrow::schema(supported_fields),
+                                             /*partition_keys=*/{}, /*primary_keys=*/{}, options));
+    ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+
+    arrow::FieldVector unsupported_fields = {
+        arrow::field("row", arrow::struct_({arrow::field("value", arrow::int32())})),
+        VariantTypeUtils::ToArrowField("variant"),
+        arrow::field("vector", arrow::fixed_size_list(arrow::float32(), 3)),
+        arrow::field("nested_row",
+                     arrow::list(arrow::struct_({arrow::field("value", arrow::int32())}))),
+    };
+    std::vector<std::string> expected_errors = {"type ROW", "type VARIANT", "type VECTOR",
+                                                "type ROW"};
+    for (size_t i = 0; i < unsupported_fields.size(); ++i) {
+        SCOPED_TRACE("field=" + unsupported_fields[i]->name());
+        ASSERT_OK_AND_ASSIGN(
+            table_schema,
+            TableSchema::Create(/*schema_id=*/0, arrow::schema({unsupported_fields[i]}),
+                                /*partition_keys=*/{}, /*primary_keys=*/{}, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            expected_errors[i]);
+    }
+
+    std::shared_ptr<arrow::Field> blob_field = BlobUtils::ToArrowField("blob", false);
+    std::map<std::string, std::string> blob_options = {
+        {Options::BUCKET, "-1"},
+        {Options::FILE_FORMAT, "mosaic"},
+        {Options::ROW_TRACKING_ENABLED, "true"},
+        {Options::DATA_EVOLUTION_ENABLED, "true"},
+    };
+    ASSERT_OK_AND_ASSIGN(
+        table_schema,
+        TableSchema::Create(/*schema_id=*/0,
+                            arrow::schema({arrow::field("id", arrow::int32()), blob_field}),
+                            /*partition_keys=*/{}, /*primary_keys=*/{}, blob_options));
+    ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+
+    blob_options[Options::BLOB_DESCRIPTOR_FIELD] = "blob";
+    ASSERT_OK_AND_ASSIGN(
+        table_schema,
+        TableSchema::Create(/*schema_id=*/0,
+                            arrow::schema({arrow::field("id", arrow::int32()), blob_field}),
+                            /*partition_keys=*/{}, /*primary_keys=*/{}, blob_options));
+    ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema), "type BLOB");
+}
+#endif
+
 TEST(SchemaValidationTest, TestRowTracking) {
     auto f0 = arrow::field("f0", arrow::utf8());
     auto f1 = arrow::field("f1", arrow::int32());

src/paimon/format/mosaic/mosaic_file_batch_reader.cpp

@@ -72,45 +72,34 @@ Result<std::unique_ptr<MosaicFileBatchReader>> MosaicFileBatchReader::Create(
     input_file.ctx = input_context.get();
     input_file.read_at_fn = MosaicInputContext::ReadAt;
     input_file.length_fn = MosaicInputContext::Length;
-    MosaicReaderHandle* reader = mosaic_reader_open(input_file);
+    std::unique_ptr<MosaicReaderHandle, decltype(&mosaic_reader_free)> reader(
+        mosaic_reader_open(input_file), mosaic_reader_free);
     if (reader == nullptr) {
         return MosaicFfiError("open Mosaic reader", input_context->GetCallbackStatus());
     }
 
     ::ArrowSchema ffi_schema = {};
-    if (mosaic_reader_export_schema(reader, &ffi_schema) != 0) {
-        Status status = MosaicFfiError("read Mosaic schema", input_context->GetCallbackStatus());
-        mosaic_reader_free(reader);
-        return status;
+    if (mosaic_reader_export_schema(reader.get(), &ffi_schema) != 0) {
+        return MosaicFfiError("read Mosaic schema", input_context->GetCallbackStatus());
     }
-    arrow::Result<std::shared_ptr<arrow::Schema>> schema_result = arrow::ImportSchema(&ffi_schema);
-    if (!schema_result.ok()) {
-        mosaic_reader_free(reader);
-        return ToPaimonStatus(schema_result.status());
-    }
-    std::shared_ptr<arrow::Schema> file_schema = std::move(schema_result).ValueOrDie();
+    PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Schema> file_schema,
+                                      arrow::ImportSchema(&ffi_schema));
 
     uint32_t num_row_groups = 0;
-    if (mosaic_reader_num_row_groups(reader, &num_row_groups) != 0) {
-        Status status =
-            MosaicFfiError("read Mosaic row group count", input_context->GetCallbackStatus());
-        mosaic_reader_free(reader);
-        return status;
+    if (mosaic_reader_num_row_groups(reader.get(), &num_row_groups) != 0) {
+        return MosaicFfiError("read Mosaic row group count", input_context->GetCallbackStatus());
     }
     uint64_t total_rows = 0;
     for (uint32_t row_group = 0; row_group < num_row_groups; ++row_group) {
         uint32_t row_count = 0;
-        if (mosaic_reader_row_group_num_rows(reader, row_group, &row_count) != 0) {
-            Status status =
-                MosaicFfiError("read Mosaic row count", input_context->GetCallbackStatus());
-            mosaic_reader_free(reader);
-            return status;
+        if (mosaic_reader_row_group_num_rows(reader.get(), row_group, &row_count) != 0) {
+            return MosaicFfiError("read Mosaic row count", input_context->GetCallbackStatus());
         }
         total_rows += row_count;
     }
     return std::unique_ptr<MosaicFileBatchReader>(
-        new MosaicFileBatchReader(input, batch_size, std::move(input_context), reader, file_schema,
-                                  num_row_groups, total_rows, pool, arrow_pool));
+        new MosaicFileBatchReader(input, batch_size, std::move(input_context), reader.release(),
+                                  file_schema, num_row_groups, total_rows, pool, arrow_pool));
 }
 
 MosaicFileBatchReader::~MosaicFileBatchReader() {

src/paimon/format/mosaic/mosaic_file_format_test.cpp

@@ -200,6 +200,39 @@ TEST_F(MosaicFileFormatTest, WriteThenRead) {
     AssertReadWithBatchSizes(path, schema, expected, {1, 2, 3, 5, 8});
 }
 
+TEST_F(MosaicFileFormatTest, EmptyProjectionPreservesRowCount) {
+    std::string path = PathUtil::JoinPath(directory_->Str(), "empty-projection.mosaic");
+    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> data =
+        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields),
+                                                  R"([[1,"one"],[2,"two"],[3,"three"]])")
+            .ValueOrDie();
+    ASSERT_OK(WriteFile(path, schema, data, /*batch_size=*/2));
+
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReaderBuilder> reader_builder,
+                         format_->CreateReaderBuilder(/*batch_size=*/2));
+    reader_builder->WithMemoryPool(pool_);
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> input, file_system_->Open(path));
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileBatchReader> reader, reader_builder->Build(input));
+    ::ArrowSchema ffi_schema = {};
+    ASSERT_TRUE(arrow::ExportSchema(*arrow::schema({}), &ffi_schema).ok());
+    ASSERT_OK(reader->SetReadSchema(&ffi_schema, /*predicate=*/nullptr,
+                                    /*selection_bitmap=*/std::nullopt));
+
+    for (int64_t expected_rows : {2, 1}) {
+        ASSERT_OK_AND_ASSIGN(BatchReader::ReadBatch batch, reader->NextBatch());
+        ASSERT_FALSE(BatchReader::IsEofBatch(batch));
+        std::shared_ptr<arrow::RecordBatch> record_batch =
+            arrow::ImportRecordBatch(batch.first.get(), batch.second.get()).ValueOrDie();
+        ASSERT_EQ(record_batch->num_columns(), 0);
+        ASSERT_EQ(record_batch->num_rows(), expected_rows);
+    }
+    ASSERT_OK_AND_ASSIGN(BatchReader::ReadBatch eof_batch, reader->NextBatch());
+    ASSERT_TRUE(BatchReader::IsEofBatch(eof_batch));
+}
+
 TEST_F(MosaicFileFormatTest, SetReadSchemaResetsReaderToFirstRow) {
     std::string path = PathUtil::JoinPath(directory_->Str(), "reset-reader.mosaic");
     arrow::FieldVector fields = {arrow::field("id", arrow::int32(), false),

src/paimon/format/mosaic/mosaic_format_writer.cpp

@@ -66,6 +66,9 @@ Result<std::unique_ptr<MosaicFormatWriter>> MosaicFormatWriter::Create(
 
 MosaicFormatWriter::~MosaicFormatWriter() {
     if (writer_ != nullptr) {
+        if (!finished_) {
+            mosaic_writer_close(writer_);
+        }
         mosaic_writer_free(writer_);
     }
 }

test/inte/blob_table_inte_test.cpp

@@ -3426,6 +3426,9 @@ TEST_P(BlobTableInteTest, TestDataEvolutionWithBlobDescriptorField) {
 }
 
 TEST_P(BlobTableInteTest, TestBlobDescriptorFieldWriteRawBytesDirectly) {
+    if (GetParam() == "mosaic") {
+        return;
+    }
     // Similar to TestBlobDescriptorField but writes raw bytes directly without converting to
     // descriptor first. Descriptor fields reject values without the descriptor magic header.
     arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),