fix(dataset)!: conflict a NOT NULL tightening with concurrent writes

java/lance-jni/src/transaction.rs

@@ -470,13 +470,19 @@ fn convert_to_java_operation_inner<'local>(
                 ],
             )?)
         }
-        Operation::Project { schema } => {
+        Operation::Project {
+            schema,
+            preserves_nullability,
+        } => {
             let java_schema = convert_to_java_schema(env, schema)?;
 
             Ok(env.new_object(
                 "org/lance/operation/Project",
-                "(Lorg/apache/arrow/vector/types/pojo/Schema;)V",
-                &[JValue::Object(&java_schema)],
+                "(Lorg/apache/arrow/vector/types/pojo/Schema;Z)V",
+                &[
+                    JValue::Object(&java_schema),
+                    JValue::Bool(preserves_nullability as u8),
+                ],
             )?)
         }
         Operation::Rewrite {
@@ -552,16 +558,18 @@ fn convert_to_java_operation_inner<'local>(
         Operation::Merge {
             fragments: rust_fragments,
             schema,
+            preserves_nullability,
         } => {
             let java_fragments = export_vec(env, &rust_fragments)?;
             let java_schema = convert_to_java_schema(env, schema)?;
 
             Ok(env.new_object(
                 "org/lance/operation/Merge",
-                "(Ljava/util/List;Lorg/apache/arrow/vector/types/pojo/Schema;)V",
+                "(Ljava/util/List;Lorg/apache/arrow/vector/types/pojo/Schema;Z)V",
                 &[
                     JValue::Object(&java_fragments),
                     JValue::Object(&java_schema),
+                    JValue::Bool(preserves_nullability as u8),
                 ],
             )?)
         }
@@ -1044,6 +1052,8 @@ fn convert_to_rust_operation(
     let op_name = env.get_string_from_method(java_operation, "name")?;
     let op = match op_name.as_str() {
         "Project" => Operation::Project {
+            preserves_nullability: env
+                .get_boolean_from_method(java_operation, "preservesNullability")?,
             schema: convert_schema_from_operation(
                 env,
                 java_operation,
@@ -1283,6 +1293,8 @@ fn convert_to_rust_operation(
                 })?;
             Operation::Merge {
                 fragments,
+                preserves_nullability: env
+                    .get_boolean_from_method(java_operation, "preservesNullability")?,
                 schema: convert_schema_from_operation(
                     env,
                     java_operation,

java/src/main/java/org/lance/operation/Merge.java

@@ -27,16 +27,26 @@
  */
 public class Merge extends SchemaOperation {
   private final List<FragmentMetadata> fragments;
+  // True when this merge makes no nullability-affecting schema change: it
+  // introduces no field that data staged against an earlier schema could not
+  // safely omit. Without the assertion the merge conservatively conflicts with
+  // concurrent appends, whose fragments would omit new columns and read as null.
+  private final boolean preservesNullability;
 
-  protected Merge(List<FragmentMetadata> fragments, Schema schema) {
+  protected Merge(List<FragmentMetadata> fragments, Schema schema, boolean preservesNullability) {
     super(schema);
     this.fragments = fragments;
+    this.preservesNullability = preservesNullability;
   }
 
   public List<FragmentMetadata> fragments() {
     return fragments;
   }
 
+  public boolean preservesNullability() {
+    return preservesNullability;
+  }
+
   @Override
   public String name() {
     return "Merge";
@@ -47,6 +57,7 @@ public String toString() {
     return MoreObjects.toStringHelper(this)
         .add("fragments", fragments)
         .add("schema", schema())
+        .add("preservesNullability", preservesNullability)
         .toString();
   }
 
@@ -56,12 +67,13 @@ public boolean equals(Object o) {
     if (o == null || getClass() != o.getClass()) return false;
     if (!super.equals(o)) return false;
     Merge that = (Merge) o;
-    return Objects.equals(fragments, that.fragments);
+    return Objects.equals(fragments, that.fragments)
+        && preservesNullability == that.preservesNullability;
   }
 
   @Override
   public int hashCode() {
-    return Objects.hash(super.hashCode(), fragments);
+    return Objects.hash(super.hashCode(), fragments, preservesNullability);
   }
 
   public static Builder builder() {
@@ -71,6 +83,8 @@ public static Builder builder() {
   public static class Builder {
     private List<FragmentMetadata> fragments;
     private Schema schema;
+    // No assertion by default, which conservatively conflicts.
+    private boolean preservesNullability = false;
 
     private Builder() {}
 
@@ -84,8 +98,13 @@ public Builder schema(Schema schema) {
       return this;
     }
 
+    public Builder preservesNullability(boolean preservesNullability) {
+      this.preservesNullability = preservesNullability;
+      return this;
+    }
+
     public Merge build() {
-      return new Merge(fragments, schema);
+      return new Merge(fragments, schema, preservesNullability);
     }
   }
 }

java/src/main/java/org/lance/operation/Project.java

@@ -16,14 +16,26 @@
 import com.google.common.base.MoreObjects;
 import org.apache.arrow.vector.types.pojo.Schema;
 
+import java.util.Objects;
+
 /**
  * Project to a new schema. This Operation only changes the schema, not the data. Note: 1. For
  * removing columns. The data will be removed after compaction. 2. Project will modify column
  * positions, not ids(a.k.a. field id)
  */
 public class Project extends SchemaOperation {
-  private Project(Schema schema) {
+  // True when this projection makes no nullability-affecting schema change,
+  // as a rename or a drop does not. Without the assertion the projection
+  // conservatively conflicts with concurrent value writes.
+  private final boolean preservesNullability;
+
+  private Project(Schema schema, boolean preservesNullability) {
     super(schema);
+    this.preservesNullability = preservesNullability;
+  }
+
+  public boolean preservesNullability() {
+    return preservesNullability;
   }
 
   @Override
@@ -33,7 +45,23 @@ public String name() {
 
   @Override
   public String toString() {
-    return MoreObjects.toStringHelper(this).add("schema", schema()).toString();
+    return MoreObjects.toStringHelper(this)
+        .add("schema", schema())
+        .add("preservesNullability", preservesNullability)
+        .toString();
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (!super.equals(o)) {
+      return false;
+    }
+    return preservesNullability == ((Project) o).preservesNullability;
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(schema(), preservesNullability);
   }
 
   public static Builder builder() {
@@ -42,6 +70,7 @@ public static Builder builder() {
 
   public static class Builder {
     private Schema schema;
+    private boolean preservesNullability;
 
     public Builder() {}
 
@@ -50,8 +79,13 @@ public Builder schema(Schema schema) {
       return this;
     }
 
+    public Builder preservesNullability(boolean preservesNullability) {
+      this.preservesNullability = preservesNullability;
+      return this;
+    }
+
     public Project build() {
-      return new Project(schema);
+      return new Project(schema, preservesNullability);
     }
   }
 }

java/src/test/java/org/lance/operation/MergeTest.java

@@ -104,10 +104,14 @@ void testMergeNewColumn(@TempDir Path tempDir) throws Exception {
                       Merge.builder()
                           .fragments(Collections.singletonList(evolvedFragment))
                           .schema(evolvedSchema)
+                          .preservesNullability(true)
                           .build())
                   .build()) {
             try (Dataset evolvedDataset = new CommitBuilder(initialDataset).execute(mergeTxn)) {
               Assertions.assertEquals(3, evolvedDataset.version());
+              // The explicit non-default assertion must survive the JNI round trip.
+              Transaction readBack = evolvedDataset.readTransaction().orElseThrow();
+              Assertions.assertTrue(((Merge) readBack.operation()).preservesNullability());
               Assertions.assertEquals(rowCount, evolvedDataset.countRows());
               Assertions.assertEquals(evolvedSchema, evolvedDataset.getSchema());
               Assertions.assertEquals(3, evolvedDataset.getSchema().getFields().size());
@@ -136,6 +140,27 @@ void testMergeNewColumn(@TempDir Path tempDir) throws Exception {
     }
   }
 
+  @Test
+  void testPreservesNullabilityEquality() {
+    Schema schema =
+        new Schema(
+            Collections.singletonList(Field.nullable("id", new ArrowType.Int(32, true))), null);
+    // No assertion by default, and the assertion is part of the operation's identity.
+    Assertions.assertFalse(
+        Merge.builder()
+            .fragments(Collections.emptyList())
+            .schema(schema)
+            .build()
+            .preservesNullability());
+    Assertions.assertNotEquals(
+        Merge.builder().fragments(Collections.emptyList()).schema(schema).build(),
+        Merge.builder()
+            .fragments(Collections.emptyList())
+            .schema(schema)
+            .preservesNullability(true)
+            .build());
+  }
+
   @Test
   void testMergeNewColumnWithNonContiguousFieldId(@TempDir Path tempDir) throws Exception {
     String datasetPath = tempDir.resolve("testMergeNewColumnWithNonContiguousFieldId").toString();

java/src/test/java/org/lance/operation/ProjectTest.java

@@ -30,6 +30,8 @@
 import java.util.List;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 public class ProjectTest extends OperationTestBase {
 
@@ -69,4 +71,36 @@ void testProjection(@TempDir Path tempDir) {
       }
     }
   }
+
+  @Test
+  void testPreservesNullabilityEqualityAndRoundTrip(@TempDir Path tempDir) {
+    String datasetPath = tempDir.resolve("testAssertsNonNull").toString();
+    try (RootAllocator allocator = new RootAllocator(Long.MAX_VALUE)) {
+      TestUtils.SimpleTestDataset testDataset =
+          new TestUtils.SimpleTestDataset(allocator, datasetPath);
+      dataset = testDataset.createEmptyDataset();
+      Schema schema = testDataset.getSchema();
+
+      // The assertion is part of the operation's identity.
+      assertNotEquals(
+          Project.builder().schema(schema).preservesNullability(true).build(),
+          Project.builder().schema(schema).preservesNullability(false).build());
+      assertEquals(
+          Project.builder().schema(schema).preservesNullability(true).build(),
+          Project.builder().schema(schema).preservesNullability(true).build());
+
+      // The explicit non-default assertion must survive the JNI round trip.
+      try (Transaction txn =
+          new Transaction.Builder()
+              .readVersion(dataset.version())
+              .operation(Project.builder().schema(schema).preservesNullability(true).build())
+              .build()) {
+        try (Dataset committed = new CommitBuilder(dataset).execute(txn)) {
+          Transaction readBack = committed.readTransaction().orElseThrow();
+          Project project = (Project) readBack.operation();
+          assertTrue(project.preservesNullability());
+        }
+      }
+    }
+  }
 }

protos/transaction.proto

@@ -144,12 +144,24 @@ message Transaction {
     repeated lance.file.Field schema = 2;
     // Schema metadata.
     map<string, bytes> schema_metadata = 3;
+    // Set when this merge makes no nullability-affecting schema change: it
+    // introduces no field that data staged against an earlier schema could
+    // not safely omit. Without the assertion (including transactions written
+    // before this field existed) the merge conservatively conflicts with
+    // concurrent value-writes, which can only cause a retry.
+    bool preserves_nullability = 4;
   }
 
   // An operation that projects a subset of columns, altering the schema.
   message Project {
     // The new schema
     repeated lance.file.Field schema = 1;
+    // Set when this projection makes no nullability-affecting schema change,
+    // as a rename or a drop does not. Without the assertion (including
+    // transactions written before this field existed) the projection
+    // conservatively conflicts with concurrent value-writes, which can only
+    // cause a retry. A nullability tightening must not set this.
+    bool preserves_nullability = 2;
   }
 
   // An operation that restores a dataset to a previous version.

python/python/lance/dataset.py

@@ -6032,6 +6032,14 @@ class Merge(BaseOperation):
         schema: LanceSchema or pyarrow.Schema
             The schema of the new dataset. Passing a LanceSchema is preferred,
             and passing a pyarrow.Schema is deprecated.
+        preserves_nullability: bool
+            True when this merge makes no nullability-affecting schema change:
+            it introduces no field that data staged against an earlier schema
+            could not safely omit. Without the assertion (the default) the
+            merge conservatively conflicts with concurrent appends, whose
+            fragments would omit new columns and read as null; that can only
+            cause a retry. Pass True when every column this merge introduces
+            is nullable to let concurrent appends commit without conflict.
 
         Warning
         -------
@@ -6077,6 +6085,7 @@ class Merge(BaseOperation):
 
         fragments: Iterable[FragmentMetadata]
         schema: LanceSchema | pa.Schema
+        preserves_nullability: bool = False
 
         def __post_init__(self):
             if isinstance(self.schema, pa.Schema):
@@ -6249,6 +6258,11 @@ class Project(BaseOperation):
         ----------
         schema: LanceSchema
             The lance schema of the new dataset.
+        preserves_nullability: bool
+            True when this projection makes no nullability-affecting schema
+            change, as a rename or a drop does not. Without the assertion
+            (the default) the projection conservatively conflicts with
+            concurrent writes, which can only cause a retry.
 
         Examples
         --------
@@ -6277,6 +6291,7 @@ class Project(BaseOperation):
         """
 
         schema: LanceSchema
+        preserves_nullability: bool = False
 
     @dataclass
     class UpdateMap: