feat: remove support for FLOAT/DOUBLE type partition field (#333)

Author: lszskye

src/paimon/core/io/field_mapping_reader.cpp

@@ -220,16 +220,6 @@ Result<std::shared_ptr<arrow::Array>> FieldMappingReader::GenerateSinglePartitio
             scalar = std::make_shared<arrow::Int64Scalar>(value);
             break;
         }
-        case arrow::Type::type::FLOAT: {
-            float value = partition_.GetFloat(partition_info_.value().idx_in_partition[idx]);
-            scalar = std::make_shared<arrow::FloatScalar>(value);
-            break;
-        }
-        case arrow::Type::type::DOUBLE: {
-            double value = partition_.GetDouble(partition_info_.value().idx_in_partition[idx]);
-            scalar = std::make_shared<arrow::DoubleScalar>(value);
-            break;
-        }
         case arrow::Type::type::STRING: {
             BinaryString value =
                 partition_.GetString(partition_info_.value().idx_in_partition[idx]);

src/paimon/core/io/field_mapping_reader_test.cpp

@@ -209,104 +209,85 @@ class FieldMappingReaderTest : public ::testing::Test {
 
 TEST_F(FieldMappingReaderTest, TestGenerateSinglePartitionArray) {
     PartitionInfo partition_info;
-    // read schema: p9-p0
-    // partition key: p0-p9
-    partition_info.partition_read_schema = {DataField(9, arrow::field("p9", arrow::date32())),
-                                            DataField(8, arrow::field("p8", arrow::binary())),
-                                            DataField(7, arrow::field("p7", arrow::utf8())),
-                                            DataField(6, arrow::field("p6", arrow::float64())),
-                                            DataField(5, arrow::field("p5", arrow::float32())),
+    // read schema: p7-p0
+    // partition key: p0-p7
+    partition_info.partition_read_schema = {DataField(7, arrow::field("p7", arrow::date32())),
+                                            DataField(6, arrow::field("p6", arrow::binary())),
+                                            DataField(5, arrow::field("p5", arrow::utf8())),
                                             DataField(4, arrow::field("p4", arrow::int64())),
                                             DataField(3, arrow::field("p3", arrow::int32())),
                                             DataField(2, arrow::field("p2", arrow::int16())),
                                             DataField(1, arrow::field("p1", arrow::int8())),
                                             DataField(0, arrow::field("p0", arrow::boolean()))};
-    partition_info.idx_in_target_read_schema = {0, 1, 2, 3, 4, 5, 6, 7, 8, 9};
-    partition_info.idx_in_partition = {9, 8, 7, 6, 5, 4, 3, 2, 1, 0};
+    partition_info.idx_in_target_read_schema = {0, 1, 2, 3, 4, 5, 6, 7};
+    partition_info.idx_in_partition = {7, 6, 5, 4, 3, 2, 1, 0};
 
     NonPartitionInfo non_part_info;
     auto field_mapping = std::make_unique<FieldMapping>(partition_info, non_part_info,
                                                         /*non_exist_field_info=*/std::nullopt);
     auto partition = BinaryRowGenerator::GenerateRow(
         {false, static_cast<int8_t>(1), static_cast<int16_t>(2), static_cast<int32_t>(3),
-         static_cast<int64_t>(4), static_cast<float>(5.1), 6.21, std::string("7"),
-         std::make_shared<Bytes>("8", pool_.get()), 100},
+         static_cast<int64_t>(4), std::string("5"), std::make_shared<Bytes>("6", pool_.get()), 100},
         pool_.get());
     auto mapping_reader = std::make_unique<FieldMappingReader>(
-        /*field_count=*/10, /*reader=*/nullptr, partition, std::move(field_mapping), pool_);
+        /*field_count=*/8, /*reader=*/nullptr, partition, std::move(field_mapping), pool_);
 
-    {
-        ASSERT_OK_AND_ASSIGN(auto p9_array, mapping_reader->GenerateSinglePartitionArray(
-                                                /*idx in read schema*/ 0, /*batch_size=*/2));
-        ASSERT_EQ(p9_array->length(), 2);
-        ASSERT_EQ(arrow::internal::checked_cast<arrow::Date32Array*>(p9_array.get())->Value(0),
-                  100);
-    }
-    {
-        ASSERT_OK_AND_ASSIGN(auto p8_array, mapping_reader->GenerateSinglePartitionArray(
-                                                /*idx in read schema*/ 1, /*batch_size=*/2));
-        ASSERT_EQ(p8_array->length(), 2);
-        ASSERT_EQ(arrow::internal::checked_cast<arrow::BinaryArray*>(p8_array.get())->Value(0),
-                  "8");
-    }
     {
         ASSERT_OK_AND_ASSIGN(auto p7_array, mapping_reader->GenerateSinglePartitionArray(
-                                                /*idx in read schema*/ 2, /*batch_size=*/1));
-        ASSERT_EQ(p7_array->length(), 1);
-        ASSERT_EQ(arrow::internal::checked_cast<arrow::StringArray*>(p7_array.get())->Value(0),
-                  "7");
+                                                /*idx=*/0, /*batch_size=*/2));
+        ASSERT_EQ(p7_array->length(), 2);
+        ASSERT_EQ(arrow::internal::checked_cast<arrow::Date32Array*>(p7_array.get())->Value(0),
+                  100);
     }
     {
         ASSERT_OK_AND_ASSIGN(auto p6_array, mapping_reader->GenerateSinglePartitionArray(
-                                                /*idx in read schema*/ 3, /*batch_size=*/1));
-        ASSERT_EQ(
-            arrow::internal::checked_cast<arrow::NumericArray<arrow::DoubleType>*>(p6_array.get())
-                ->Value(0),
-            static_cast<double>(6.21));
+                                                /*idx=*/1, /*batch_size=*/2));
+        ASSERT_EQ(p6_array->length(), 2);
+        ASSERT_EQ(arrow::internal::checked_cast<arrow::BinaryArray*>(p6_array.get())->Value(0),
+                  "6");
     }
     {
         ASSERT_OK_AND_ASSIGN(auto p5_array, mapping_reader->GenerateSinglePartitionArray(
-                                                /*idx in read schema*/ 4, /*batch_size=*/1));
-        ASSERT_EQ(
-            arrow::internal::checked_cast<arrow::NumericArray<arrow::FloatType>*>(p5_array.get())
-                ->Value(0),
-            static_cast<float>(5.1));
+                                                /*idx=*/2, /*batch_size=*/1));
+        ASSERT_EQ(p5_array->length(), 1);
+        ASSERT_EQ(arrow::internal::checked_cast<arrow::StringArray*>(p5_array.get())->Value(0),
+                  "5");
     }
     {
         ASSERT_OK_AND_ASSIGN(auto p4_array, mapping_reader->GenerateSinglePartitionArray(
-                                                /*idx in read schema*/ 5, /*batch_size=*/1));
+                                                /*idx=*/3, /*batch_size=*/1));
         ASSERT_EQ(
             arrow::internal::checked_cast<arrow::NumericArray<arrow::Int64Type>*>(p4_array.get())
                 ->Value(0),
             static_cast<int64_t>(4));
     }
     {
         ASSERT_OK_AND_ASSIGN(auto p3_array, mapping_reader->GenerateSinglePartitionArray(
-                                                /*idx in read schema*/ 6, /*batch_size=*/1));
+                                                /*idx=*/4, /*batch_size=*/1));
         ASSERT_EQ(
             arrow::internal::checked_cast<arrow::NumericArray<arrow::Int32Type>*>(p3_array.get())
                 ->Value(0),
             static_cast<int32_t>(3));
     }
     {
         ASSERT_OK_AND_ASSIGN(auto p2_array, mapping_reader->GenerateSinglePartitionArray(
-                                                /*idx in read schema*/ 7, /*batch_size=*/1));
+                                                /*idx=*/5, /*batch_size=*/1));
         ASSERT_EQ(
             arrow::internal::checked_cast<arrow::NumericArray<arrow::Int16Type>*>(p2_array.get())
                 ->Value(0),
             static_cast<int16_t>(2));
     }
     {
         ASSERT_OK_AND_ASSIGN(auto p1_array, mapping_reader->GenerateSinglePartitionArray(
-                                                /*idx in read schema*/ 8, /*batch_size=*/1));
+                                                /*idx=*/6, /*batch_size=*/1));
         ASSERT_EQ(
             arrow::internal::checked_cast<arrow::NumericArray<arrow::Int8Type>*>(p1_array.get())
                 ->Value(0),
             static_cast<int64_t>(1));
     }
     {
         ASSERT_OK_AND_ASSIGN(auto p0_array, mapping_reader->GenerateSinglePartitionArray(
-                                                /*idx in read schema*/ 9, /*batch_size=*/1));
+                                                /*idx=*/7, /*batch_size=*/1));
         ASSERT_EQ(arrow::internal::checked_cast<arrow::BooleanArray*>(p0_array.get())->Value(0),
                   false);
     }

src/paimon/core/schema/schema_validation.cpp

@@ -73,9 +73,9 @@ Status SchemaValidation::ValidateTableSchema(const TableSchema& schema) {
         ValidateOnlyContainPrimitiveType(schema.Fields(), schema.PrimaryKeys(), "primary key"));
     PAIMON_RETURN_NOT_OK(
         ValidateOnlyContainPrimitiveType(schema.Fields(), schema.PartitionKeys(), "partition"));
-    // TODO(lisizhuo.lsz): C++ Paimon do not support timestamp & decimal type in partition keys for
-    // now.
-    PAIMON_RETURN_NOT_OK(ValidateNotContainComplexType(schema.Fields(), schema.PartitionKeys()));
+    // TODO(lisizhuo.lsz): C++ Paimon do not support timestamp & decimal & float & double type in
+    // partition keys for now.
+    PAIMON_RETURN_NOT_OK(ValidateNotContainSpecificType(schema.Fields(), schema.PartitionKeys()));
 
     PAIMON_ASSIGN_OR_RAISE(CoreOptions options, CoreOptions::FromMap(schema.Options()));
     PAIMON_RETURN_NOT_OK(ValidateBucket(schema, options));
@@ -163,7 +163,7 @@ Status SchemaValidation::ValidateOnlyContainPrimitiveType(
     return Status::OK();
 }
 
-Status SchemaValidation::ValidateNotContainComplexType(
+Status SchemaValidation::ValidateNotContainSpecificType(
     const std::vector<DataField>& fields, const std::vector<std::string>& field_names) {
     if (field_names.empty()) {
         return Status::OK();
@@ -178,8 +178,12 @@ Status SchemaValidation::ValidateNotContainComplexType(
             auto field = it->second;
             if (IsComplexType(field)) {
                 return Status::Invalid(
-                    fmt::format("The field {} in partition field {} is unsupported",
-                                field->ToString(), it->first));
+                    fmt::format("partition field {} cannot be TIMESTAMP/DECIMAL/BLOB", field_name));
+            }
+            if (field->type()->id() == arrow::Type::FLOAT ||
+                field->type()->id() == arrow::Type::DOUBLE) {
+                return Status::Invalid(
+                    fmt::format("partition field {} cannot be FLOAT/DOUBLE", field_name));
             }
         } else {
             assert(false);

src/paimon/core/schema/schema_validation.h

@@ -54,8 +54,8 @@ class SchemaValidation {
     static Status ValidateOnlyContainPrimitiveType(const std::vector<DataField>& fields,
                                                    const std::vector<std::string>& field_names,
                                                    const std::string& error_message_intro);
-    static Status ValidateNotContainComplexType(const std::vector<DataField>& fields,
-                                                const std::vector<std::string>& field_names);
+    static Status ValidateNotContainSpecificType(const std::vector<DataField>& fields,
+                                                 const std::vector<std::string>& field_names);
     static Status ValidateBucket(const TableSchema& schema, const CoreOptions& options);
     static Status ValidateDefaultValues(const TableSchema& schema) {
         return Status::NotImplemented("validate default values not implemented");

src/paimon/core/schema/schema_validation_test.cpp

@@ -393,19 +393,34 @@ TEST(SchemaValidationTest, NonPrimitivePartitionKeyStruct) {
                         "field f1 is unsupported");
 }
 
-TEST(SchemaValidationTest, TestComplexPartitionKey) {
-    auto f0 = arrow::field("f0", arrow::utf8());
-    auto f1 = arrow::field("f1", arrow::decimal128(5, 2));
-    auto f2 = arrow::field("f2", arrow::float64());
-    arrow::FieldVector fields = {f0, f1, f2};
-    auto schema = arrow::schema(fields);
-    std::vector<std::string> primary_keys = {"f0", "f1"};
-    std::vector<std::string> partition_keys = {"f1"};
-    ASSERT_OK_AND_ASSIGN(
-        std::shared_ptr<TableSchema> table_schema,
-        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, {}));
-    ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
-                        "partition field f1 is unsupported");
+TEST(SchemaValidationTest, TestSpecificPartitionKey) {
+    {
+        auto f0 = arrow::field("f0", arrow::utf8());
+        auto f1 = arrow::field("f1", arrow::decimal128(5, 2));
+        auto f2 = arrow::field("f2", arrow::float64());
+        arrow::FieldVector fields = {f0, f1, f2};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {"f0", "f1"};
+        std::vector<std::string> partition_keys = {"f1"};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, {}));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "partition field f1 cannot be TIMESTAMP/DECIMAL/BLOB");
+    }
+    {
+        auto f0 = arrow::field("f0", arrow::utf8());
+        auto f1 = arrow::field("f1", arrow::float64());
+        arrow::FieldVector fields = {f0, f1};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {"f0", "f1"};
+        std::vector<std::string> partition_keys = {"f1"};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, {}));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "partition field f1 cannot be FLOAT/DOUBLE");
+    }
 }
 
 TEST(SchemaValidationTest, TestComplexPartitionKeyWithBlob) {
@@ -419,7 +434,7 @@ TEST(SchemaValidationTest, TestComplexPartitionKeyWithBlob) {
         std::shared_ptr<TableSchema> table_schema,
         TableSchema::Create(/*schema_id=*/0, schema, partition_keys, /*primary_keys=*/{}, {}));
     ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
-                        "partition field f1 is unsupported");
+                        "partition field f1 cannot be TIMESTAMP/DECIMAL/BLOB");
 }
 
 TEST(SchemaValidationTest, TestDateTypePartitionKey) {
@@ -511,8 +526,8 @@ TEST(SchemaValidationTest, ValidateBucket) {
                             "The number of buckets needs to be greater than 0.");
     }
     {
-        std::vector<std::string> primary_keys = {"f0", "f1"};
-        std::vector<std::string> partition_keys = {"f2"};
+        std::vector<std::string> primary_keys = {"f0", "f2"};
+        std::vector<std::string> partition_keys = {"f1"};
         std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
                                                       {Options::BUCKET_KEY, "f0"}};
         ASSERT_OK_AND_ASSIGN(
@@ -524,7 +539,7 @@ TEST(SchemaValidationTest, ValidateBucket) {
     }
     {
         std::vector<std::string> primary_keys = {};
-        std::vector<std::string> partition_keys = {"f2"};
+        std::vector<std::string> partition_keys = {"f1"};
         std::map<std::string, std::string> options = {{Options::BUCKET, "2"}};
         ASSERT_OK_AND_ASSIGN(
             std::shared_ptr<TableSchema> table_schema,
@@ -533,7 +548,7 @@ TEST(SchemaValidationTest, ValidateBucket) {
                             "You should define a 'bucket-key' for bucketed append mode");
     }
     {
-        std::vector<std::string> partition_keys = {"f2"};
+        std::vector<std::string> partition_keys = {"f1"};
         std::map<std::string, std::string> options = {{"full-compaction.delta-commits", "2"}};
         ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
                              TableSchema::Create(/*schema_id=*/0, schema, partition_keys,
@@ -633,11 +648,11 @@ TEST(SchemaValidationTest, ValidateSequenceField) {
         std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
                                                       {Options::SEQUENCE_FIELD, "f0,f1,f2"}};
         ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
-                             TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{"f2"},
-                                                 /*primary_keys=*/{"f0", "f1"}, options));
+                             TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{"f1"},
+                                                 /*primary_keys=*/{"f0", "f2"}, options));
         ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
                             "You cannot use sequence.field in cross partition update case (Primary "
-                            "key constraint 'f0, f1'  not including all partition fields 'f2').");
+                            "key constraint 'f0, f2'  not including all partition fields 'f1').");
     }
 }
 

src/paimon/core/stats/simple_stats_collector.cpp

@@ -133,40 +133,6 @@ Status SimpleStatsCollector::Collect(const BinaryRow& row) {
                 }
                 break;
             }
-            case arrow::Type::FLOAT: {
-                if (column_stats_[i] == nullptr) {
-                    column_stats_[i] = ColumnStats::CreateFloatColumnStats(
-                        std::nullopt, std::nullopt, std::nullopt);
-                }
-                auto typed_stats = dynamic_cast<FloatColumnStats*>(column_stats_[i].get());
-                if (typed_stats == nullptr) {
-                    assert(false);
-                    return Status::Invalid("cast typed stats failed");
-                }
-                if (!row.IsNullAt(i)) {
-                    typed_stats->Collect(row.GetFloat(i));
-                } else {
-                    typed_stats->Collect(std::nullopt);
-                }
-                break;
-            }
-            case arrow::Type::DOUBLE: {
-                if (column_stats_[i] == nullptr) {
-                    column_stats_[i] = ColumnStats::CreateDoubleColumnStats(
-                        std::nullopt, std::nullopt, std::nullopt);
-                }
-                auto typed_stats = dynamic_cast<DoubleColumnStats*>(column_stats_[i].get());
-                if (typed_stats == nullptr) {
-                    assert(false);
-                    return Status::Invalid("cast typed stats failed");
-                }
-                if (!row.IsNullAt(i)) {
-                    typed_stats->Collect(row.GetDouble(i));
-                } else {
-                    typed_stats->Collect(std::nullopt);
-                }
-                break;
-            }
             case arrow::Type::STRING:
             case arrow::Type::BINARY: {
                 if (column_stats_[i] == nullptr) {