[python]: Add nullability support for data type in python

bindings/python/src/metadata.rs

@@ -165,7 +165,7 @@ impl Schema {
         let mut builder = fcore::metadata::Schema::builder();
 
         for field in arrow_schema.fields() {
-            let fluss_data_type = crate::utils::Utils::arrow_type_to_fluss_type(field.data_type())?;
+            let fluss_data_type = crate::utils::Utils::arrow_field_to_fluss_type(field)?;
             builder = builder.column(field.name(), fluss_data_type);
 
             if let Some(comment) = field.metadata().get("comment") {

bindings/python/src/utils.rs

@@ -36,14 +36,14 @@ impl Utils {
         })
     }
 
-    /// Convert Arrow DataType to Fluss DataType
-    pub fn arrow_type_to_fluss_type(
-        arrow_type: &arrow::datatypes::DataType,
+    /// Convert an Arrow Field to a Fluss DataType, preserving nullability.
+    pub fn arrow_field_to_fluss_type(
+        field: &arrow::datatypes::Field,
     ) -> PyResult<fcore::metadata::DataType> {
         use arrow::datatypes::DataType as ArrowDataType;
         use fcore::metadata::DataTypes;
 
-        let fluss_type = match arrow_type {
+        let fluss_type = match field.data_type() {
             ArrowDataType::Boolean => DataTypes::boolean(),
             ArrowDataType::Int8 => DataTypes::tinyint(),
             ArrowDataType::Int16 => DataTypes::smallint(),
@@ -95,23 +95,29 @@ impl Utils {
             ArrowDataType::Decimal128(precision, scale) => {
                 DataTypes::decimal(*precision as u32, *scale as u32)
             }
-            ArrowDataType::List(field) => {
-                let element_type = Utils::arrow_type_to_fluss_type(field.data_type())?;
+            ArrowDataType::List(element_field) => {
+                let element_type = Utils::arrow_field_to_fluss_type(element_field)?;
                 DataTypes::array(element_type)
             }
-            _ => {
+            other => {
                 return Err(FlussError::new_err(format!(
-                    "Unsupported Arrow data type: {arrow_type:?}"
+                    "Unsupported Arrow data type: {other:?}"
                 )));
             }
         };
 
-        Ok(fluss_type)
+        if field.is_nullable() {
+            Ok(fluss_type)
+        } else {
+            Ok(fluss_type.as_non_nullable())
+        }
     }
 
-    /// Convert Fluss DataType to string representation
+    /// Convert Fluss DataType to string representation, appending " NOT NULL"
+    /// for non-nullable types (matches Java's `withNullability` and Rust core's
+    /// `Display` impl).
     pub fn datatype_to_string(data_type: &fcore::metadata::DataType) -> String {
-        match data_type {
+        let type_str = match data_type {
             fcore::metadata::DataType::Boolean(_) => "boolean".to_string(),
             fcore::metadata::DataType::TinyInt(_) => "tinyint".to_string(),
             fcore::metadata::DataType::SmallInt(_) => "smallint".to_string(),
@@ -171,6 +177,12 @@ impl Utils {
                     .collect();
                 format!("row<{}>", fields.join(", "))
             }
+        };
+
+        if data_type.is_nullable() {
+            type_str
+        } else {
+            format!("{type_str} NOT NULL")
         }
     }
 

bindings/python/test/test_schema.py

@@ -48,3 +48,50 @@ def test_schema_with_array():
     assert schema.get_column_types() == ["int", "array<string>"]
 
 
+def test_nullable_fields():
+    fields = pa.schema(
+        [
+            pa.field("id", pa.int32(), nullable=False),
+            pa.field("name", pa.string()),
+        ]
+    )
+    schema = fluss.Schema(fields)
+    assert schema.get_column_types() == ["int NOT NULL", "string"]
+    assert schema.get_columns() == [("id", "int NOT NULL"), ("name", "string")]
+
+
+def test_pk_forces_non_nullable():
+    fields = pa.schema(
+        [
+            pa.field("id", pa.int32()),
+            pa.field("name", pa.string()),
+        ]
+    )
+    schema = fluss.Schema(fields, primary_keys=["id"])
+    types = schema.get_column_types()
+    assert types[0] == "int NOT NULL"
+    assert types[1] == "string"
+
+
+def test_nested_list_nullability():
+    fields = pa.schema(
+        [
+            pa.field(
+                "tags",
+                pa.list_(pa.field("item", pa.string(), nullable=False)),
+            ),
+            pa.field("ids", pa.list_(pa.int32()), nullable=False),
+            pa.field(
+                "strict_ids",
+                pa.list_(pa.field("item", pa.int32(), nullable=False)),
+                nullable=False,
+            ),
+        ]
+    )
+    schema = fluss.Schema(fields)
+    types = schema.get_column_types()
+    assert types[0] == "array<string NOT NULL>"
+    assert types[1] == "array<int> NOT NULL"
+    assert types[2] == "array<int NOT NULL> NOT NULL"
+
+

website/docs/user-guide/python/api-reference.md

@@ -242,10 +242,10 @@ for record in scan_records:
 
 | Method                                         |  Description               |
 |------------------------------------------------|----------------------------|
-| `Schema(schema: pa.Schema, primary_keys=None)` | Create from PyArrow schema |
+| `Schema(schema: pa.Schema, primary_keys=None)` | Create from PyArrow schema. Field nullability (`pa.field(..., nullable=False)`) is preserved. |
 | `.get_column_names() -> list[str]`             | Get column names           |
-| `.get_column_types() -> list[str]`             | Get column type names      |
-| `.get_columns() -> list[tuple[str, str]]`      | Get `(name, type)` pairs   |
+| `.get_column_types() -> list[str]`             | Get column type names. Non-nullable types include a `" NOT NULL"` suffix (e.g., `"int NOT NULL"`). |
+| `.get_columns() -> list[tuple[str, str]]`      | Get `(name, type)` pairs. Type strings follow the same nullability formatting as `.get_column_types()`. |
 | `.get_primary_keys() -> list[str]`             | Get primary key columns    |
 
 ## `TableDescriptor`

website/docs/user-guide/python/data-types.md

@@ -21,6 +21,31 @@ The Python client uses PyArrow types for schema definitions:
 
 All Python native types (`date`, `time`, `datetime`, `Decimal`) work when appending rows via dicts.
 
+## Nullability
+
+PyArrow field nullability is preserved when constructing Fluss schemas. By default, fields are nullable. Use `nullable=False` on `pa.field()` to create a `NOT NULL` column:
+
+```python
+schema = pa.schema([
+    pa.field("id", pa.int32(), nullable=False),
+    pa.field("name", pa.string()),          # nullable by default
+])
+fluss_schema = fluss.Schema(schema)
+fluss_schema.get_column_types()  # ["int NOT NULL", "string"]
+```
+
+Primary key columns are automatically forced `NOT NULL` regardless of the PyArrow field setting.
+
+For nested types, element nullability is also preserved:
+
+```python
+schema = pa.schema([
+    pa.field("tags", pa.list_(pa.field("item", pa.string(), nullable=False))),
+])
+fluss_schema = fluss.Schema(schema)
+fluss_schema.get_column_types()  # ["array<string NOT NULL>"]
+```
+
 ## Writing Data
 
 Rows can be dicts, lists, or tuples: