[#12064] fix(lance): complete V3 type compatibility

catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/lance/LanceTableOperations.java

@@ -252,6 +252,10 @@ public Table createTable(
     Preconditions.checkArgument(
         StringUtils.isNotBlank(location), "Table location must be specified");
 
+    // Validate every field before EXIST_OK or OVERWRITE can return, drop metadata, or purge a
+    // dataset. Conversion failures must never leave a partial mutation.
+    convertColumnsToArrowSchema(columns);
+
     // Extract creation mode from properties
     CreationMode mode =
         Optional.ofNullable(properties.get(LANCE_CREATION_MODE))

catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/lance/TestLanceTableOperations.java

@@ -121,6 +121,36 @@ public void testCreateTableWithInvalidMode() {
                 new Index[0]));
   }
 
+  @Test
+  public void testVariantTypeIsRejectedBeforeOverwriteMutation() {
+    NameIdentifier ident = NameIdentifier.of("catalog", "schema", "table");
+    Column[] columns = {Column.of("payload", Types.VariantType.get(), "variant")};
+    Map<String, String> properties =
+        Map.of(
+            Table.PROPERTY_LOCATION,
+            tempDir.resolve("variant-overwrite").toString(),
+            LANCE_CREATION_MODE,
+            "OVERWRITE");
+
+    IllegalArgumentException exception =
+        Assertions.assertThrows(
+            IllegalArgumentException.class,
+            () ->
+                lanceTableOps.createTable(
+                    ident,
+                    columns,
+                    null,
+                    properties,
+                    new Transform[0],
+                    null,
+                    new SortOrder[0],
+                    new Index[0]));
+
+    Assertions.assertTrue(exception.getMessage().contains("exact native representation"));
+    verify(lanceTableOps, never()).purgeTable(ident);
+    verify(lanceTableOps, never()).dropTable(ident);
+  }
+
   @Test
   public void testLoadDeclaredTableSchemaFromLocation() throws Exception {
     NameIdentifier ident = NameIdentifier.of("schema", "table");

catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/lance/integration/test/CatalogGenericCatalogLanceIT.java

@@ -44,6 +44,7 @@
 import org.apache.arrow.vector.VarCharVector;
 import org.apache.arrow.vector.VectorSchemaRoot;
 import org.apache.arrow.vector.ipc.ArrowReader;
+import org.apache.arrow.vector.types.TimeUnit;
 import org.apache.arrow.vector.types.pojo.ArrowType;
 import org.apache.arrow.vector.types.pojo.Field;
 import org.apache.commons.io.FileUtils;
@@ -413,6 +414,203 @@ public void testCreateLanceTable() {
         RuntimeException.class, () -> catalog.asTableCatalog().loadTable(newNameIdentifier));
   }
 
+  @Test
+  public void testNanosecondTimestampTypeRoundTrip() {
+    String phase2TableName = GravitinoITUtils.genRandomName("lance_timestamp_ns");
+    NameIdentifier identifier = NameIdentifier.of(schemaName, phase2TableName);
+    String location = tempDirectory + "/" + phase2TableName;
+    Column[] columns = {
+      Column.of("timestamp_ns", Types.TimestampType.withoutTimeZone(9), "nanosecond timestamp"),
+      Column.of("timestamptz_ns", Types.TimestampType.withTimeZone(9), "nanosecond instant")
+    };
+    Map<String, String> properties = createProperties();
+    properties.put(Table.PROPERTY_TABLE_FORMAT, LANCE_TABLE_FORMAT);
+    properties.put(Table.PROPERTY_LOCATION, location);
+
+    Table created =
+        catalog
+            .asTableCatalog()
+            .createTable(
+                identifier,
+                columns,
+                "timestamp nanosecond round-trip",
+                properties,
+                Transforms.EMPTY_TRANSFORM,
+                null,
+                null);
+    Table loaded = catalog.asTableCatalog().loadTable(identifier);
+
+    Assertions.assertEquals(
+        Types.TimestampType.withoutTimeZone(9), created.columns()[0].dataType());
+    Assertions.assertEquals(Types.TimestampType.withTimeZone(9), created.columns()[1].dataType());
+    Assertions.assertEquals(Types.TimestampType.withoutTimeZone(9), loaded.columns()[0].dataType());
+    Assertions.assertEquals(Types.TimestampType.withTimeZone(9), loaded.columns()[1].dataType());
+
+    try (Dataset dataset = Dataset.open().uri(location).build()) {
+      List<Field> fields = dataset.getSchema().getFields();
+      Assertions.assertEquals(
+          new ArrowType.Timestamp(TimeUnit.NANOSECOND, null), fields.get(0).getType());
+      Assertions.assertEquals(
+          new ArrowType.Timestamp(TimeUnit.NANOSECOND, "UTC"), fields.get(1).getType());
+    }
+  }
+
+  @Test
+  public void testVariantOverwriteRejectedWithoutSideEffects() {
+    String phase2TableName = GravitinoITUtils.genRandomName("lance_variant_rejection");
+    NameIdentifier identifier = NameIdentifier.of(schemaName, phase2TableName);
+    String location = tempDirectory + "/" + phase2TableName;
+    Map<String, String> properties = createProperties();
+    properties.put(Table.PROPERTY_TABLE_FORMAT, LANCE_TABLE_FORMAT);
+    properties.put(Table.PROPERTY_LOCATION, location);
+    properties.put(Table.PROPERTY_EXTERNAL, "true");
+    Column[] originalColumns = {
+      Column.of("id", Types.IntegerType.get(), "original integer column")
+    };
+
+    catalog
+        .asTableCatalog()
+        .createTable(
+            identifier,
+            originalColumns,
+            "original table",
+            properties,
+            Transforms.EMPTY_TRANSFORM,
+            null,
+            null);
+
+    Map<String, String> overwriteProperties = Maps.newHashMap(properties);
+    overwriteProperties.put(LANCE_CREATION_MODE, "OVERWRITE");
+    IllegalArgumentException exception =
+        Assertions.assertThrows(
+            IllegalArgumentException.class,
+            () ->
+                catalog
+                    .asTableCatalog()
+                    .createTable(
+                        identifier,
+                        new Column[] {Column.of("payload", Types.VariantType.get(), "variant")},
+                        "invalid overwrite",
+                        overwriteProperties,
+                        Transforms.EMPTY_TRANSFORM,
+                        null,
+                        null));
+
+    Assertions.assertTrue(exception.getMessage().contains("exact native representation"));
+    Assertions.assertTrue(catalog.asTableCatalog().tableExists(identifier));
+    Assertions.assertEquals(
+        Types.IntegerType.get(),
+        catalog.asTableCatalog().loadTable(identifier).columns()[0].dataType());
+    try (Dataset dataset = Dataset.open().uri(location).build()) {
+      Assertions.assertEquals(
+          new ArrowType.Int(32, true), dataset.getSchema().getFields().get(0).getType());
+    }
+  }
+
+  @Test
+  public void testUnknownTypeRoundTrip() {
+    String phase2TableName = GravitinoITUtils.genRandomName("lance_unknown");
+    NameIdentifier identifier = NameIdentifier.of(schemaName, phase2TableName);
+    String location = tempDirectory + "/" + phase2TableName;
+    Column[] columns = {Column.of("unknown_value", Types.NullType.get(), "null-only value")};
+    Map<String, String> properties = createProperties();
+    properties.put(Table.PROPERTY_TABLE_FORMAT, LANCE_TABLE_FORMAT);
+    properties.put(Table.PROPERTY_LOCATION, location);
+
+    Table created =
+        catalog
+            .asTableCatalog()
+            .createTable(
+                identifier,
+                columns,
+                "unknown type round-trip",
+                properties,
+                Transforms.EMPTY_TRANSFORM,
+                null,
+                null);
+    Table loaded = catalog.asTableCatalog().loadTable(identifier);
+
+    Assertions.assertEquals(Types.NullType.get(), created.columns()[0].dataType());
+    Assertions.assertEquals(Types.NullType.get(), loaded.columns()[0].dataType());
+    Assertions.assertTrue(created.columns()[0].nullable());
+    try (Dataset dataset = Dataset.open().uri(location).build()) {
+      Field field = dataset.getSchema().getFields().get(0);
+      Assertions.assertEquals(ArrowType.Null.INSTANCE, field.getType());
+      Assertions.assertTrue(field.isNullable());
+    }
+  }
+
+  @Test
+  public void testGeometryTypeRoundTrip() {
+    String phase2TableName = GravitinoITUtils.genRandomName("lance_geometry");
+    NameIdentifier identifier = NameIdentifier.of(schemaName, phase2TableName);
+    String location = tempDirectory + "/" + phase2TableName;
+    Types.GeometryType geometry = Types.GeometryType.of("EPSG:3857");
+    Column[] columns = {Column.of("shape", geometry, "planar WKB geometry")};
+    Map<String, String> properties = createProperties();
+    properties.put(Table.PROPERTY_TABLE_FORMAT, LANCE_TABLE_FORMAT);
+    properties.put(Table.PROPERTY_LOCATION, location);
+
+    Table created =
+        catalog
+            .asTableCatalog()
+            .createTable(
+                identifier,
+                columns,
+                "geometry type round-trip",
+                properties,
+                Transforms.EMPTY_TRANSFORM,
+                null,
+                null);
+    Table loaded = catalog.asTableCatalog().loadTable(identifier);
+
+    Assertions.assertEquals(geometry, created.columns()[0].dataType());
+    Assertions.assertEquals(geometry, loaded.columns()[0].dataType());
+    try (Dataset dataset = Dataset.open().uri(location).build()) {
+      Field field = dataset.getSchema().getFields().get(0);
+      Assertions.assertEquals(ArrowType.Binary.INSTANCE, field.getType());
+      Assertions.assertEquals("geoarrow.wkb", field.getMetadata().get("ARROW:extension:name"));
+      Assertions.assertEquals(
+          "{\"crs\":\"EPSG:3857\"}", field.getMetadata().get("ARROW:extension:metadata"));
+    }
+  }
+
+  @Test
+  public void testGeographyTypeRoundTrip() {
+    String phase2TableName = GravitinoITUtils.genRandomName("lance_geography");
+    NameIdentifier identifier = NameIdentifier.of(schemaName, phase2TableName);
+    String location = tempDirectory + "/" + phase2TableName;
+    Types.GeographyType geography = Types.GeographyType.of("EPSG:4326", "karney");
+    Column[] columns = {Column.of("shape", geography, "ellipsoidal WKB geography")};
+    Map<String, String> properties = createProperties();
+    properties.put(Table.PROPERTY_TABLE_FORMAT, LANCE_TABLE_FORMAT);
+    properties.put(Table.PROPERTY_LOCATION, location);
+
+    Table created =
+        catalog
+            .asTableCatalog()
+            .createTable(
+                identifier,
+                columns,
+                "geography type round-trip",
+                properties,
+                Transforms.EMPTY_TRANSFORM,
+                null,
+                null);
+    Table loaded = catalog.asTableCatalog().loadTable(identifier);
+
+    Assertions.assertEquals(geography, created.columns()[0].dataType());
+    Assertions.assertEquals(geography, loaded.columns()[0].dataType());
+    try (Dataset dataset = Dataset.open().uri(location).build()) {
+      Field field = dataset.getSchema().getFields().get(0);
+      Assertions.assertEquals(ArrowType.Binary.INSTANCE, field.getType());
+      Assertions.assertEquals("geoarrow.wkb", field.getMetadata().get("ARROW:extension:name"));
+      Assertions.assertEquals(
+          "{\"crs\":\"EPSG:4326\",\"edges\":\"karney\"}",
+          field.getMetadata().get("ARROW:extension:metadata"));
+    }
+  }
+
   @Test
   void testLanceTableFormat() {
     String tableName = GravitinoITUtils.genRandomName(TABLE_PREFIX);

docs/lakehouse-generic-lance-table.md

@@ -70,12 +70,37 @@ Lance uses Apache Arrow for table schemas. The following table shows type mappin
 | `Timestamp_tz(3)`                | `TimestampType Millisecond withUtc`     |
 | `Timestamp_tz(9)`                | `TimestampType Nanosecond withUtc`      |
 | `Time`/`Time(9)`                 | `Time Nanosecond`                       |
-| `Null`                           | `Null`                                  |
+| `Unknown` (`NullType`)           | `Null` (nullable columns only)          |
+| `Variant`                        | Rejected before mutation                |
+| `Geometry(crs)`                  | `geoarrow.wkb` with planar CRS metadata |
+| `Geography(crs, algorithm)`      | `geoarrow.wkb` with CRS and edge metadata |
 | `Fixed(n)`                       | `Fixed-Size Binary(n)`                  |
 | `Interval_year`                  | Not supported by Lance                  |
 | `Interval_day`                   | `Duration(Microsecond)`                 |
 | `External(arrow_field_json_str)` | Any Arrow Field                         |
 
+`Timestamp(9)` and `Timestamp_tz(9)` round-trip losslessly through Lance as nanosecond Arrow
+timestamps. Gravitino `Timestamp_tz` has time-zone-aware instant semantics but does not carry a
+zone identifier, so the connector writes the canonical Arrow `UTC` identifier. Native Arrow
+timestamps with another zone identifier are preserved as `External` instead of silently rewriting
+that identifier.
+
+Lance 6.0 with Arrow 18 has no exact native representation for Gravitino `Variant`, so table
+creation rejects it before creating, replacing, or deleting table data. Arrow extension fields,
+including newer `arrow.parquet.variant` fields, remain lossless `External` types.
+
+Gravitino `Unknown` is a nullable, null-only placeholder whose concrete type may be assigned during
+schema evolution. It round-trips exactly as Arrow `Null`; a non-nullable Unknown column is rejected
+because it cannot contain a valid value.
+
+Gravitino `Geometry` uses WKB with planar edges and CRS type metadata. Lance preserves the same
+semantics as a GeoArrow 0.2 `geoarrow.wkb` extension field backed by Arrow `Binary`; both textual
+CRS identifiers and PROJJSON round-trip without losing the CRS.
+
+Gravitino `Geography` uses the same GeoArrow WKB storage and records its spherical or spheroidal
+edge algorithm in the GeoArrow `edges` metadata. All five Gravitino algorithms (`spherical`,
+`vincenty`, `thomas`, `andoyer`, and `karney`) round-trip with the CRS.
+
 ### External Types
 
 For Arrow types not natively mapped in Gravitino, use the `External(arrow_field_json_str)` type, which accepts a JSON string representation of an Arrow `Field`.