diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt
index e48209d62..3b5f86009 100644
--- a/src/iceberg/CMakeLists.txt
+++ b/src/iceberg/CMakeLists.txt
@@ -47,6 +47,7 @@ set(ICEBERG_SOURCES
     file_io_registry.cc
     file_reader.cc
     file_writer.cc
+    geospatial.cc
     inspect/history_table.cc
     inspect/metadata_table.cc
     inspect/snapshots_table.cc
diff --git a/src/iceberg/avro/avro_data_util.cc b/src/iceberg/avro/avro_data_util.cc
index 9be8907d4..297e46b52 100644
--- a/src/iceberg/avro/avro_data_util.cc
+++ b/src/iceberg/avro/avro_data_util.cc
@@ -334,7 +334,9 @@ Status AppendPrimitiveValueToBuilder(const ::avro::NodePtr& avro_node,
       return {};
     }
 
-    case TypeId::kBinary: {
+    case TypeId::kBinary:
+    case TypeId::kGeometry:
+    case TypeId::kGeography: {
       if (avro_node->type() != ::avro::AVRO_BYTES) {
         return InvalidArgument("Expected Avro bytes for binary field, got: {}",
                                ToString(avro_node));
diff --git a/src/iceberg/avro/avro_direct_decoder.cc b/src/iceberg/avro/avro_direct_decoder.cc
index 8f563b3ba..23bb8ca7f 100644
--- a/src/iceberg/avro/avro_direct_decoder.cc
+++ b/src/iceberg/avro/avro_direct_decoder.cc
@@ -465,7 +465,9 @@ Status DecodePrimitiveValueToBuilder(const ::avro::NodePtr& avro_node,
       return {};
     }
 
-    case TypeId::kBinary: {
+    case TypeId::kBinary:
+    case TypeId::kGeometry:
+    case TypeId::kGeography: {
       if (avro_node->type() != ::avro::AVRO_BYTES) {
         return InvalidArgument("Expected Avro bytes for binary field, got: {}",
                                ToString(avro_node));
diff --git a/src/iceberg/avro/avro_schema_util.cc b/src/iceberg/avro/avro_schema_util.cc
index 5e6bee957..5b10fd05e 100644
--- a/src/iceberg/avro/avro_schema_util.cc
+++ b/src/iceberg/avro/avro_schema_util.cc
@@ -252,12 +252,14 @@ Status ToAvroNodeVisitor::Visit(const VariantType&, ::avro::NodePtr*) {
   return NotSupported("Writing Iceberg variant type to Avro is not supported");
 }
 
-Status ToAvroNodeVisitor::Visit(const GeometryType&, ::avro::NodePtr*) {
-  return NotSupported("Writing Iceberg geometry type to Avro is not supported");
+Status ToAvroNodeVisitor::Visit(const GeometryType&, ::avro::NodePtr* node) {
+  *node = std::make_shared<::avro::NodePrimitive>(::avro::AVRO_BYTES);
+  return {};
 }
 
-Status ToAvroNodeVisitor::Visit(const GeographyType&, ::avro::NodePtr*) {
-  return NotSupported("Writing Iceberg geography type to Avro is not supported");
+Status ToAvroNodeVisitor::Visit(const GeographyType&, ::avro::NodePtr* node) {
+  *node = std::make_shared<::avro::NodePrimitive>(::avro::AVRO_BYTES);
+  return {};
 }
 
 Status ToAvroNodeVisitor::Visit(const StructType& type, ::avro::NodePtr* node) {
@@ -637,6 +639,8 @@ Status ValidateAvroSchemaEvolution(const Type& expected_type,
       }
       break;
     case TypeId::kBinary:
+    case TypeId::kGeometry:
+    case TypeId::kGeography:
       if (avro_node->type() == ::avro::AVRO_BYTES) {
         return {};
       }
@@ -644,8 +648,6 @@ Status ValidateAvroSchemaEvolution(const Type& expected_type,
     case TypeId::kUnknown:
       return {};
     case TypeId::kVariant:
-    case TypeId::kGeometry:
-    case TypeId::kGeography:
       return NotSupported("Reading Iceberg type {} from Avro is not supported",
                           expected_type);
     default:
diff --git a/src/iceberg/geospatial.cc b/src/iceberg/geospatial.cc
new file mode 100644
index 000000000..db034d178
--- /dev/null
+++ b/src/iceberg/geospatial.cc
@@ -0,0 +1,269 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/geospatial.h"
+
+#include <cmath>
+#include <cstddef>
+#include <cstdint>
+#include <limits>
+#include <optional>
+#include <span>
+#include <utility>
+#include <vector>
+
+#include "iceberg/type.h"
+#include "iceberg/util/endian.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+namespace {
+
+constexpr size_t kDoubleSize = sizeof(double);
+
+std::optional<double> OptionalOrdinate(double value) {
+  return std::isnan(value) ? std::nullopt : std::make_optional(value);
+}
+
+template <EndianConvertible T>
+void AppendLittleEndian(std::vector<uint8_t>& bytes, T value) {
+  const size_t offset = bytes.size();
+  bytes.resize(offset + sizeof(value));
+  WriteLittleEndian(value, bytes.data() + offset);
+}
+
+bool IsValidBoundSize(size_t size) {
+  return size == 2 * kDoubleSize || size == 3 * kDoubleSize || size == 4 * kDoubleSize;
+}
+
+bool RangeIntersects(double min1, double max1, double min2, double max2) {
+  return min1 <= max2 && max1 >= min2;
+}
+
+bool RangeIntersectsWithWrapAround(double min1, double max1, double min2, double max2) {
+  // A wrapped longitude interval [min, max], where min > max, represents
+  // [min, 180] plus [-180, max].
+  const bool interval1_wraps = min1 > max1;
+  const bool interval2_wraps = min2 > max2;
+  if (!interval1_wraps && !interval2_wraps) {
+    return RangeIntersects(min1, max1, min2, max2);
+  }
+  if (interval1_wraps && interval2_wraps) {
+    // Both intervals contain the antimeridian, so they always intersect.
+    return true;
+  }
+  if (interval1_wraps) {
+    return min1 <= max2 || max1 >= min2;
+  }
+  return min2 <= max1 || max2 >= min1;
+}
+
+Status ValidateOptionalRanges(const BoundingBox& bbox) {
+  if (bbox.lower().z().has_value() && bbox.upper().z().has_value() &&
+      *bbox.lower().z() > *bbox.upper().z()) {
+    return InvalidArgument("Invalid Z range: zmin cannot be greater than zmax");
+  }
+  if (bbox.lower().m().has_value() && bbox.upper().m().has_value() &&
+      *bbox.lower().m() > *bbox.upper().m()) {
+    return InvalidArgument("Invalid M range: mmin cannot be greater than mmax");
+  }
+  return {};
+}
+
+Status ValidateGeometry(const BoundingBox& bbox) {
+  if (!(bbox.lower().x() <= bbox.upper().x())) {
+    return InvalidArgument("Invalid X range: xmin cannot be greater than xmax");
+  }
+  if (!(bbox.lower().y() <= bbox.upper().y())) {
+    return InvalidArgument("Invalid Y range: ymin cannot be greater than ymax");
+  }
+  return ValidateOptionalRanges(bbox);
+}
+
+Status ValidateGeography(const BoundingBox& bbox) {
+  if (!(bbox.lower().y() >= -90.0 && bbox.lower().y() <= 90.0 &&
+        bbox.upper().y() >= -90.0 && bbox.upper().y() <= 90.0)) {
+    return InvalidArgument("Invalid latitude: out of range [-90, 90]");
+  }
+  if (!(bbox.lower().x() >= -180.0 && bbox.lower().x() <= 180.0 &&
+        bbox.upper().x() >= -180.0 && bbox.upper().x() <= 180.0)) {
+    return InvalidArgument("Invalid longitude: out of range [-180, 180]");
+  }
+  if (bbox.lower().y() > bbox.upper().y()) {
+    return InvalidArgument("Invalid latitude range: ymin cannot be greater than ymax");
+  }
+  return ValidateOptionalRanges(bbox);
+}
+
+bool IntersectsYzm(const BoundingBox& lhs, const BoundingBox& rhs) {
+  if (lhs.lower().z().has_value() && lhs.upper().z().has_value() &&
+      rhs.lower().z().has_value() && rhs.upper().z().has_value() &&
+      !RangeIntersects(*lhs.lower().z(), *lhs.upper().z(), *rhs.lower().z(),
+                       *rhs.upper().z())) {
+    return false;
+  }
+  if (lhs.lower().m().has_value() && lhs.upper().m().has_value() &&
+      rhs.lower().m().has_value() && rhs.upper().m().has_value() &&
+      !RangeIntersects(*lhs.lower().m(), *lhs.upper().m(), *rhs.lower().m(),
+                       *rhs.upper().m())) {
+    return false;
+  }
+  return RangeIntersects(lhs.lower().y(), lhs.upper().y(), rhs.lower().y(),
+                         rhs.upper().y());
+}
+
+}  // namespace
+
+GeospatialBound::GeospatialBound(double x, double y, std::optional<double> z,
+                                 std::optional<double> m)
+    : x_(x), y_(y), z_(std::move(z)), m_(std::move(m)) {}
+
+GeospatialBound GeospatialBound::XY(double x, double y) {
+  return GeospatialBound{x, y, std::nullopt, std::nullopt};
+}
+
+GeospatialBound GeospatialBound::XYZ(double x, double y, double z) {
+  return GeospatialBound{x, y, OptionalOrdinate(z), std::nullopt};
+}
+
+GeospatialBound GeospatialBound::XYM(double x, double y, double m) {
+  return GeospatialBound{x, y, std::nullopt, OptionalOrdinate(m)};
+}
+
+GeospatialBound GeospatialBound::XYZM(double x, double y, double z, double m) {
+  return GeospatialBound{x, y, OptionalOrdinate(z), OptionalOrdinate(m)};
+}
+
+Result<GeospatialBound> GeospatialBound::Deserialize(std::span<const uint8_t> bytes) {
+  if (!IsValidBoundSize(bytes.size())) {
+    return InvalidArgument(
+        "Invalid geospatial bound size: {}. Valid sizes are 16, 24, or 32 bytes",
+        bytes.size());
+  }
+
+  auto x = ReadLittleEndian<double>(bytes.data());
+  auto y = ReadLittleEndian<double>(bytes.data() + kDoubleSize);
+  if (bytes.size() == 2 * kDoubleSize) {
+    return XY(x, y);
+  }
+
+  auto z = ReadLittleEndian<double>(bytes.data() + 2 * kDoubleSize);
+  if (bytes.size() == 3 * kDoubleSize) {
+    return XYZ(x, y, z);
+  }
+  auto m = ReadLittleEndian<double>(bytes.data() + 3 * kDoubleSize);
+  if (std::isnan(z)) {
+    return XYM(x, y, m);
+  }
+  return XYZM(x, y, z, m);
+}
+
+std::vector<uint8_t> GeospatialBound::Serialize() const {
+  size_t size = 2 * kDoubleSize;
+  if (z_.has_value() && !m_.has_value()) {
+    size = 3 * kDoubleSize;
+  } else if (m_.has_value()) {
+    size = 4 * kDoubleSize;
+  }
+
+  std::vector<uint8_t> bytes;
+  bytes.reserve(size);
+  AppendLittleEndian(bytes, x_);
+  AppendLittleEndian(bytes, y_);
+  if (z_.has_value() || m_.has_value()) {
+    AppendLittleEndian(bytes, z_.value_or(std::numeric_limits<double>::quiet_NaN()));
+  }
+  if (m_.has_value()) {
+    AppendLittleEndian(bytes, *m_);
+  }
+  return bytes;
+}
+
+BoundingBox::BoundingBox(GeospatialBound lower, GeospatialBound upper)
+    : lower_(std::move(lower)), upper_(std::move(upper)) {}
+
+Result<BoundingBox> BoundingBox::Deserialize(std::span<const uint8_t> lower,
+                                             std::span<const uint8_t> upper) {
+  ICEBERG_ASSIGN_OR_RAISE(auto lower_bound, GeospatialBound::Deserialize(lower));
+  ICEBERG_ASSIGN_OR_RAISE(auto upper_bound, GeospatialBound::Deserialize(upper));
+  return BoundingBox(std::move(lower_bound), std::move(upper_bound));
+}
+
+Result<BoundingBox> BoundingBox::Deserialize(std::span<const uint8_t> bytes) {
+  constexpr size_t kLengthSize = sizeof(uint32_t);
+  if (bytes.size() < kLengthSize) {
+    return InvalidArgument("Truncated bounding box encoding");
+  }
+  const size_t lower_size = ReadLittleEndian<uint32_t>(bytes.data());
+  if (!IsValidBoundSize(lower_size) ||
+      bytes.size() < kLengthSize + lower_size + kLengthSize) {
+    return InvalidArgument("Invalid or truncated lower geospatial bound");
+  }
+
+  const size_t upper_length_offset = kLengthSize + lower_size;
+  const size_t upper_size =
+      ReadLittleEndian<uint32_t>(bytes.data() + upper_length_offset);
+  const size_t upper_offset = upper_length_offset + kLengthSize;
+  if (!IsValidBoundSize(upper_size) || bytes.size() < upper_offset + upper_size) {
+    return InvalidArgument("Invalid or truncated upper geospatial bound");
+  }
+
+  return Deserialize(bytes.subspan(kLengthSize, lower_size),
+                     bytes.subspan(upper_offset, upper_size));
+}
+
+std::vector<uint8_t> BoundingBox::Serialize() const {
+  auto lower_bytes = lower_.Serialize();
+  auto upper_bytes = upper_.Serialize();
+  std::vector<uint8_t> bytes;
+  bytes.reserve(sizeof(uint32_t) + lower_bytes.size() + sizeof(uint32_t) +
+                upper_bytes.size());
+  AppendLittleEndian(bytes, static_cast<uint32_t>(lower_bytes.size()));
+  bytes.insert(bytes.end(), lower_bytes.begin(), lower_bytes.end());
+  AppendLittleEndian(bytes, static_cast<uint32_t>(upper_bytes.size()));
+  bytes.insert(bytes.end(), upper_bytes.begin(), upper_bytes.end());
+  return bytes;
+}
+
+Result<bool> GeospatialBoundsIntersect(const Type& type, const BoundingBox& lhs,
+                                       const BoundingBox& rhs) {
+  switch (type.type_id()) {
+    case TypeId::kGeometry:
+      ICEBERG_RETURN_UNEXPECTED(ValidateGeometry(lhs));
+      ICEBERG_RETURN_UNEXPECTED(ValidateGeometry(rhs));
+      if (!IntersectsYzm(lhs, rhs)) {
+        return false;
+      }
+      return RangeIntersects(lhs.lower().x(), lhs.upper().x(), rhs.lower().x(),
+                             rhs.upper().x());
+    case TypeId::kGeography:
+      ICEBERG_RETURN_UNEXPECTED(ValidateGeography(lhs));
+      ICEBERG_RETURN_UNEXPECTED(ValidateGeography(rhs));
+      if (!IntersectsYzm(lhs, rhs)) {
+        return false;
+      }
+      return RangeIntersectsWithWrapAround(lhs.lower().x(), lhs.upper().x(),
+                                           rhs.lower().x(), rhs.upper().x());
+    default:
+      return NotSupported("Unsupported type for BoundingBox: {}", type.ToString());
+  }
+}
+
+}  // namespace iceberg
diff --git a/src/iceberg/geospatial.h b/src/iceberg/geospatial.h
new file mode 100644
index 000000000..9c991afb1
--- /dev/null
+++ b/src/iceberg/geospatial.h
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#pragma once
+
+/// \file iceberg/geospatial.h
+/// \brief Geospatial bounds and intersection utilities.
+
+#include <cstdint>
+#include <optional>
+#include <span>
+#include <vector>
+
+#include "iceberg/iceberg_export.h"
+#include "iceberg/result.h"
+#include "iceberg/type_fwd.h"
+
+namespace iceberg {
+
+/// \brief A point used as a lower or upper geospatial column bound.
+class ICEBERG_EXPORT GeospatialBound {
+ public:
+  /// \brief Create a two-dimensional bound.
+  static GeospatialBound XY(double x, double y);
+  /// \brief Create a bound with a Z coordinate.
+  static GeospatialBound XYZ(double x, double y, double z);
+  /// \brief Create a bound with an M coordinate.
+  static GeospatialBound XYM(double x, double y, double m);
+  /// \brief Create a bound with Z and M coordinates.
+  static GeospatialBound XYZM(double x, double y, double z, double m);
+
+  /// \brief Return the X coordinate.
+  double x() const { return x_; }
+  /// \brief Return the Y coordinate.
+  double y() const { return y_; }
+  /// \brief Return the optional Z coordinate.
+  const std::optional<double>& z() const { return z_; }
+  /// \brief Return the optional M coordinate.
+  const std::optional<double>& m() const { return m_; }
+
+  /// \brief Decode a little-endian geospatial bound.
+  static Result<GeospatialBound> Deserialize(std::span<const uint8_t> bytes);
+  /// \brief Encode this bound using little-endian encoding.
+  std::vector<uint8_t> Serialize() const;
+
+  friend bool operator==(const GeospatialBound&, const GeospatialBound&) = default;
+
+ private:
+  GeospatialBound(double x, double y, std::optional<double> z, std::optional<double> m);
+
+  double x_;
+  double y_;
+  std::optional<double> z_;
+  std::optional<double> m_;
+};
+
+/// \brief A pair of lower and upper geospatial column bounds.
+class ICEBERG_EXPORT BoundingBox {
+ public:
+  /// \brief Create a bounding box from lower and upper bounds.
+  BoundingBox(GeospatialBound lower, GeospatialBound upper);
+
+  /// \brief Return the lower bound.
+  const GeospatialBound& lower() const { return lower_; }
+  /// \brief Return the upper bound.
+  const GeospatialBound& upper() const { return upper_; }
+
+  /// \brief Decode a bounding box from separate lower and upper encodings.
+  static Result<BoundingBox> Deserialize(std::span<const uint8_t> lower,
+                                         std::span<const uint8_t> upper);
+  /// \brief Decode a bounding box from Iceberg's combined encoding.
+  static Result<BoundingBox> Deserialize(std::span<const uint8_t> bytes);
+  /// \brief Encode this bounding box using Iceberg's combined encoding.
+  std::vector<uint8_t> Serialize() const;
+
+  friend bool operator==(const BoundingBox&, const BoundingBox&) = default;
+
+ private:
+  GeospatialBound lower_;
+  GeospatialBound upper_;
+};
+
+/// \brief Check whether two bounding boxes intersect for an Iceberg geospatial type.
+///
+/// Returns an error for invalid boxes or a non-geospatial type.
+ICEBERG_EXPORT Result<bool> GeospatialBoundsIntersect(const Type& type,
+                                                      const BoundingBox& lhs,
+                                                      const BoundingBox& rhs);
+
+}  // namespace iceberg
diff --git a/src/iceberg/meson.build b/src/iceberg/meson.build
index 39e8b2939..422f97558 100644
--- a/src/iceberg/meson.build
+++ b/src/iceberg/meson.build
@@ -95,6 +95,7 @@ iceberg_sources = files(
     'file_io_registry.cc',
     'file_reader.cc',
     'file_writer.cc',
+    'geospatial.cc',
     'inheritable_metadata.cc',
     'inspect/history_table.cc',
     'inspect/metadata_table.cc',
@@ -316,6 +317,7 @@ install_headers(
         'file_io_registry.h',
         'file_reader.h',
         'file_writer.h',
+        'geospatial.h',
         'iceberg_data_export.h',
         'iceberg_export.h',
         'inheritable_metadata.h',
diff --git a/src/iceberg/parquet/parquet_metrics.cc b/src/iceberg/parquet/parquet_metrics.cc
index dac9ea6a8..22710c137 100644
--- a/src/iceberg/parquet/parquet_metrics.cc
+++ b/src/iceberg/parquet/parquet_metrics.cc
@@ -183,6 +183,10 @@ bool IsFloatingType(const PrimitiveType& type) {
   return type.type_id() == TypeId::kFloat || type.type_id() == TypeId::kDouble;
 }
 
+bool IsGeospatialType(const PrimitiveType& type) {
+  return type.type_id() == TypeId::kGeometry || type.type_id() == TypeId::kGeography;
+}
+
 bool NeedsBoundTruncation(const PrimitiveType& type) {
   return type.type_id() == TypeId::kString || type.type_id() == TypeId::kBinary;
 }
@@ -388,7 +392,7 @@ Result<std::optional<FieldMetrics>> MetricsFromFooter(
     return std::nullopt;
   }
 
-  if (truncate_length <= 0) {
+  if (truncate_length <= 0 || IsGeospatialType(*iceberg_type)) {
     return CollectCounts(field_id, metadata, column_idx.value());
   }
 
diff --git a/src/iceberg/parquet/parquet_schema_util.cc b/src/iceberg/parquet/parquet_schema_util.cc
index 658880814..b03bf106a 100644
--- a/src/iceberg/parquet/parquet_schema_util.cc
+++ b/src/iceberg/parquet/parquet_schema_util.cc
@@ -60,6 +60,193 @@ std::optional<int32_t> GetFieldId(const ::parquet::arrow::SchemaField& parquet_f
   return FieldIdFromMetadata(parquet_field.field->metadata());
 }
 
+Result<::parquet::LogicalType::EdgeInterpolationAlgorithm> ToParquetAlgorithm(
+    EdgeAlgorithm algorithm) {
+  using ParquetAlgorithm = ::parquet::LogicalType::EdgeInterpolationAlgorithm;
+  switch (algorithm) {
+    case EdgeAlgorithm::kSpherical:
+      return ParquetAlgorithm::SPHERICAL;
+    case EdgeAlgorithm::kVincenty:
+      return ParquetAlgorithm::VINCENTY;
+    case EdgeAlgorithm::kThomas:
+      return ParquetAlgorithm::THOMAS;
+    case EdgeAlgorithm::kAndoyer:
+      return ParquetAlgorithm::ANDOYER;
+    case EdgeAlgorithm::kKarney:
+      return ParquetAlgorithm::KARNEY;
+  }
+  return InvalidArgument("Unknown Iceberg edge algorithm");
+}
+
+Result<::parquet::schema::NodePtr> ToParquetNode(const SchemaField& field);
+
+Result<::parquet::schema::NodePtr> ToParquetPrimitive(
+    const PrimitiveType& primitive, std::string name,
+    ::parquet::Repetition::type repetition, int32_t field_id) {
+  using ParquetType = ::parquet::Type;
+  using TimeUnit = ::parquet::LogicalType::TimeUnit;
+
+  auto logical_type = ::parquet::LogicalType::None();
+  auto physical_type = ParquetType::UNDEFINED;
+  int32_t type_length = -1;
+
+  switch (primitive.type_id()) {
+    case TypeId::kBoolean:
+      physical_type = ParquetType::BOOLEAN;
+      break;
+    case TypeId::kInt:
+      physical_type = ParquetType::INT32;
+      break;
+    case TypeId::kLong:
+      physical_type = ParquetType::INT64;
+      break;
+    case TypeId::kFloat:
+      physical_type = ParquetType::FLOAT;
+      break;
+    case TypeId::kDouble:
+      physical_type = ParquetType::DOUBLE;
+      break;
+    case TypeId::kDecimal: {
+      const auto& decimal = internal::checked_cast<const DecimalType&>(primitive);
+      if (decimal.precision() <= 9) {
+        physical_type = ParquetType::INT32;
+      } else if (decimal.precision() <= 18) {
+        physical_type = ParquetType::INT64;
+      } else {
+        physical_type = ParquetType::FIXED_LEN_BYTE_ARRAY;
+        type_length = ::arrow::DecimalType::DecimalSize(decimal.precision());
+      }
+      logical_type =
+          ::parquet::LogicalType::Decimal(decimal.precision(), decimal.scale());
+    } break;
+    case TypeId::kDate:
+      physical_type = ParquetType::INT32;
+      logical_type = ::parquet::LogicalType::Date();
+      break;
+    case TypeId::kTime:
+      physical_type = ParquetType::INT64;
+      logical_type =
+          ::parquet::LogicalType::Time(/*is_adjusted_to_utc=*/false, TimeUnit::MICROS);
+      break;
+    case TypeId::kTimestamp:
+      physical_type = ParquetType::INT64;
+      logical_type = ::parquet::LogicalType::Timestamp(/*is_adjusted_to_utc=*/false,
+                                                       TimeUnit::MICROS);
+      break;
+    case TypeId::kTimestampTz:
+      physical_type = ParquetType::INT64;
+      logical_type = ::parquet::LogicalType::Timestamp(/*is_adjusted_to_utc=*/true,
+                                                       TimeUnit::MICROS);
+      break;
+    case TypeId::kTimestampNs:
+      physical_type = ParquetType::INT64;
+      logical_type = ::parquet::LogicalType::Timestamp(/*is_adjusted_to_utc=*/false,
+                                                       TimeUnit::NANOS);
+      break;
+    case TypeId::kTimestampTzNs:
+      physical_type = ParquetType::INT64;
+      logical_type =
+          ::parquet::LogicalType::Timestamp(/*is_adjusted_to_utc=*/true, TimeUnit::NANOS);
+      break;
+    case TypeId::kString:
+      physical_type = ParquetType::BYTE_ARRAY;
+      logical_type = ::parquet::LogicalType::String();
+      break;
+    case TypeId::kUuid:
+      physical_type = ParquetType::FIXED_LEN_BYTE_ARRAY;
+      type_length = 16;
+      logical_type = ::parquet::LogicalType::UUID();
+      break;
+    case TypeId::kFixed: {
+      const auto& fixed = internal::checked_cast<const FixedType&>(primitive);
+      physical_type = ParquetType::FIXED_LEN_BYTE_ARRAY;
+      type_length = fixed.length();
+    } break;
+    case TypeId::kBinary:
+      physical_type = ParquetType::BYTE_ARRAY;
+      break;
+    case TypeId::kGeometry: {
+      const auto& geometry = internal::checked_cast<const GeometryType&>(primitive);
+      physical_type = ParquetType::BYTE_ARRAY;
+      logical_type = ::parquet::LogicalType::Geometry(std::string(geometry.crs()));
+    } break;
+    case TypeId::kGeography: {
+      const auto& geography = internal::checked_cast<const GeographyType&>(primitive);
+      ICEBERG_ASSIGN_OR_RAISE(auto algorithm, ToParquetAlgorithm(geography.algorithm()));
+      physical_type = ParquetType::BYTE_ARRAY;
+      logical_type =
+          ::parquet::LogicalType::Geography(std::string(geography.crs()), algorithm);
+    } break;
+    case TypeId::kUnknown:
+      if (repetition != ::parquet::Repetition::OPTIONAL) {
+        return InvalidSchema("Iceberg unknown type must be optional");
+      }
+      physical_type = ParquetType::INT32;
+      logical_type = ::parquet::LogicalType::Null();
+      break;
+    case TypeId::kVariant:
+    case TypeId::kStruct:
+    case TypeId::kList:
+    case TypeId::kMap:
+      return NotSupported("Cannot write Iceberg type {} to Parquet", primitive);
+  }
+
+  return ::parquet::schema::PrimitiveNode::Make(name, repetition, std::move(logical_type),
+                                                physical_type, type_length, field_id);
+}
+
+Result<::parquet::schema::NodePtr> ToParquetNode(const SchemaField& field) {
+  const auto repetition = field.optional() ? ::parquet::Repetition::OPTIONAL
+                                           : ::parquet::Repetition::REQUIRED;
+  const auto& type = *field.type();
+  if (type.is_primitive()) {
+    return ToParquetPrimitive(internal::checked_cast<const PrimitiveType&>(type),
+                              std::string(field.name()), repetition, field.field_id());
+  }
+
+  switch (type.type_id()) {
+    case TypeId::kStruct: {
+      const auto& struct_type = internal::checked_cast<const StructType&>(type);
+      if (struct_type.fields().empty()) {
+        return NotImplemented("Cannot write empty struct '{}' to Parquet", field.name());
+      }
+      ::parquet::schema::NodeVector children;
+      children.reserve(struct_type.fields().size());
+      for (const auto& child_field : struct_type.fields()) {
+        ICEBERG_ASSIGN_OR_RAISE(auto child, ToParquetNode(child_field));
+        children.push_back(std::move(child));
+      }
+      return ::parquet::schema::GroupNode::Make(
+          std::string(field.name()), repetition, std::move(children),
+          /*logical_type=*/nullptr, field.field_id());
+    }
+    case TypeId::kList: {
+      const auto& list_type = internal::checked_cast<const ListType&>(type);
+      ICEBERG_ASSIGN_OR_RAISE(auto element, ToParquetNode(list_type.element()));
+      auto repeated = ::parquet::schema::GroupNode::Make(
+          "list", ::parquet::Repetition::REPEATED, {std::move(element)});
+      return ::parquet::schema::GroupNode::Make(
+          std::string(field.name()), repetition, {std::move(repeated)},
+          ::parquet::LogicalType::List(), field.field_id());
+    }
+    case TypeId::kMap: {
+      const auto& map_type = internal::checked_cast<const MapType&>(type);
+      ICEBERG_ASSIGN_OR_RAISE(auto key, ToParquetNode(map_type.key()));
+      ICEBERG_ASSIGN_OR_RAISE(auto value, ToParquetNode(map_type.value()));
+      auto repeated =
+          ::parquet::schema::GroupNode::Make("key_value", ::parquet::Repetition::REPEATED,
+                                             {std::move(key), std::move(value)});
+      return ::parquet::schema::GroupNode::Make(
+          std::string(field.name()), repetition, {std::move(repeated)},
+          ::parquet::LogicalType::Map(), field.field_id());
+    }
+    case TypeId::kVariant:
+      return NotSupported("Cannot write Iceberg variant type to Parquet");
+    default:
+      return InvalidSchema("Expected nested Iceberg type, got {}", type);
+  }
+}
+
 bool IsNullPhysicalField(const ::parquet::arrow::SchemaField& parquet_field) {
   return parquet_field.field->type()->id() == ::arrow::Type::NA;
 }
@@ -105,16 +292,39 @@ void SelectAnchorColumnIfEmpty(
   }
 }
 
+Status ValidateGeospatialParquetType(const Type& expected_type,
+                                     const ::parquet::arrow::SchemaField& parquet_field,
+                                     const ::parquet::ColumnDescriptor& descr) {
+  if (parquet_field.field->type()->id() != ::arrow::Type::BINARY) {
+    return InvalidSchema("Cannot read Iceberg type {} from non-binary Parquet field",
+                         expected_type);
+  }
+  if (descr.logical_type() == nullptr) {
+    return InvalidSchema("Missing Parquet leaf column metadata for Iceberg type {}",
+                         expected_type);
+  }
+  if (descr.physical_type() != ::parquet::Type::BYTE_ARRAY) {
+    return InvalidSchema("Iceberg type {} requires Parquet BYTE_ARRAY", expected_type);
+  }
+  if (expected_type.type_id() == TypeId::kGeometry) {
+    if (!descr.logical_type()->is_geometry()) {
+      return InvalidSchema("Iceberg geometry requires Parquet Geometry logical type");
+    }
+    return {};
+  }
+  if (!descr.logical_type()->is_geography()) {
+    return InvalidSchema("Iceberg geography requires Parquet Geography logical type");
+  }
+  return {};
+}
+
 }  // namespace
 
-Status ValidateParquetSchemaEvolution(
+namespace {
+
+Status ValidateParquetTypeCompatibility(
     const Type& expected_type, const ::parquet::arrow::SchemaField& parquet_field) {
   const auto& arrow_type = parquet_field.field->type();
-  // Some Parquet files may contain null-only physical fields. Allow reading them as
-  // any optional projected field type.
-  if (arrow_type->id() == ::arrow::Type::NA) {
-    return {};
-  }
   switch (expected_type.type_id()) {
     case TypeId::kBoolean:
       if (arrow_type->id() == ::arrow::Type::BOOL) {
@@ -241,8 +451,6 @@ Status ValidateParquetSchemaEvolution(
     case TypeId::kUnknown:
       return {};
     case TypeId::kVariant:
-    case TypeId::kGeometry:
-    case TypeId::kGeography:
       return NotSupported("Reading Iceberg type {} from Parquet is not supported",
                           expected_type);
     case TypeId::kStruct:
@@ -268,203 +476,228 @@ Status ValidateParquetSchemaEvolution(
                        expected_type, arrow_type->ToString());
 }
 
+}  // namespace
+
 namespace {
 
-// Forward declaration
-Result<FieldProjection> ProjectNested(
-    const Type& nested_type,
-    const std::vector<::parquet::arrow::SchemaField>& parquet_fields);
+class ProjectionBuilder {
+ public:
+  explicit ProjectionBuilder(const ::parquet::SchemaDescriptor* descr) : descr_(descr) {}
 
-Result<FieldProjection> ProjectField(const SchemaField& expected_field,
-                                     const ::parquet::arrow::SchemaField& parquet_field,
-                                     size_t source_index) {
-  const Type& expected_type = *expected_field.type();
-
-  FieldProjection projection;
-  if (expected_type.type_id() == TypeId::kUnknown || IsNullPhysicalField(parquet_field)) {
-    if (!expected_field.optional()) {
-      return InvalidSchema("Cannot project required field with id {} as null",
-                           expected_field.field_id());
-    }
-    projection.kind = FieldProjection::Kind::kNull;
-    return projection;
+  Result<FieldProjection> Project(
+      const Type& nested_type,
+      const std::vector<::parquet::arrow::SchemaField>& parquet_fields) {
+    return ProjectNested(nested_type, parquet_fields);
   }
 
-  ICEBERG_RETURN_UNEXPECTED(ValidateParquetSchemaEvolution(expected_type, parquet_field));
-
-  if (expected_type.is_nested()) {
-    ICEBERG_ASSIGN_OR_RAISE(projection,
-                            ProjectNested(expected_type, parquet_field.children));
-  } else {
-    projection.attributes =
-        std::make_shared<ParquetExtraAttributes>(parquet_field.column_index);
-  }
-  projection.from = source_index;
-  projection.kind = FieldProjection::Kind::kProjected;
-  return projection;
-}
+ private:
+  Result<FieldProjection> ProjectField(const SchemaField& expected_field,
+                                       const ::parquet::arrow::SchemaField& parquet_field,
+                                       size_t source_index) {
+    const Type& expected_type = *expected_field.type();
+
+    FieldProjection projection;
+    if (expected_type.type_id() == TypeId::kUnknown ||
+        IsNullPhysicalField(parquet_field)) {
+      if (!expected_field.optional()) {
+        return InvalidSchema("Cannot project required field with id {} as null",
+                             expected_field.field_id());
+      }
+      projection.kind = FieldProjection::Kind::kNull;
+      return projection;
+    }
 
-Result<FieldProjection> ProjectStruct(
-    const StructType& struct_type,
-    const std::vector<::parquet::arrow::SchemaField>& parquet_fields) {
-  struct FieldContext {
-    size_t local_index;
-    const ::parquet::arrow::SchemaField& parquet_field;
-  };
-  std::unordered_map<int32_t, FieldContext> field_context_map;
-  field_context_map.reserve(parquet_fields.size());
-
-  for (size_t i = 0; i < parquet_fields.size(); ++i) {
-    const ::parquet::arrow::SchemaField& parquet_field = parquet_fields[i];
-    auto field_id = GetFieldId(parquet_field);
-    if (!field_id) {
-      continue;
+    if (expected_type.type_id() == TypeId::kGeometry ||
+        expected_type.type_id() == TypeId::kGeography) {
+      if (descr_ == nullptr || parquet_field.column_index < 0 ||
+          parquet_field.column_index >= descr_->num_columns()) {
+        return InvalidSchema("Missing Parquet leaf column metadata for Iceberg type {}",
+                             expected_type);
+      }
+      ICEBERG_RETURN_UNEXPECTED(ValidateGeospatialParquetType(
+          expected_type, parquet_field, *descr_->Column(parquet_field.column_index)));
+    } else {
+      ICEBERG_RETURN_UNEXPECTED(
+          ValidateParquetTypeCompatibility(expected_type, parquet_field));
     }
-    if (!field_context_map
-             .emplace(field_id.value(),
-                      FieldContext{.local_index = i, .parquet_field = parquet_field})
-             .second) [[unlikely]] {
-      return InvalidSchema("Duplicate field id {} found in Parquet schema",
-                           field_id.value());
+
+    if (expected_type.is_nested()) {
+      ICEBERG_ASSIGN_OR_RAISE(projection,
+                              ProjectNested(expected_type, parquet_field.children));
+    } else {
+      projection.attributes =
+          std::make_shared<ParquetExtraAttributes>(parquet_field.column_index);
     }
+    projection.from = source_index;
+    projection.kind = FieldProjection::Kind::kProjected;
+    return projection;
   }
 
-  FieldProjection result;
-  result.children.reserve(struct_type.fields().size());
-
-  for (const auto& field : struct_type.fields()) {
-    int32_t field_id = field.field_id();
-    FieldProjection child_projection;
-
-    if (auto iter = field_context_map.find(field_id); iter != field_context_map.cend()) {
-      const auto& parquet_field = iter->second.parquet_field;
-      ICEBERG_ASSIGN_OR_RAISE(
-          child_projection, ProjectField(field, parquet_field, iter->second.local_index));
-    } else if (MetadataColumns::IsMetadataColumn(field_id)) {
-      child_projection.kind = FieldProjection::Kind::kMetadata;
-    } else if (field.initial_default() != nullptr) {
-      // Rows written before the field existed assume its `initial-default` value.
-      child_projection.kind = FieldProjection::Kind::kDefault;
-      child_projection.from = *field.initial_default();
-    } else if (field.optional()) {
-      child_projection.kind = FieldProjection::Kind::kNull;
-    } else {
-      return InvalidSchema("Missing required field with id: {}", field_id);
+  Result<FieldProjection> ProjectStruct(
+      const StructType& struct_type,
+      const std::vector<::parquet::arrow::SchemaField>& parquet_fields) {
+    struct FieldContext {
+      size_t local_index;
+      const ::parquet::arrow::SchemaField& parquet_field;
+    };
+    std::unordered_map<int32_t, FieldContext> field_context_map;
+    field_context_map.reserve(parquet_fields.size());
+
+    for (size_t i = 0; i < parquet_fields.size(); ++i) {
+      const ::parquet::arrow::SchemaField& parquet_field = parquet_fields[i];
+      auto field_id = GetFieldId(parquet_field);
+      if (!field_id) {
+        continue;
+      }
+      if (!field_context_map
+               .emplace(field_id.value(),
+                        FieldContext{.local_index = i, .parquet_field = parquet_field})
+               .second) [[unlikely]] {
+        return InvalidSchema("Duplicate field id {} found in Parquet schema",
+                             field_id.value());
+      }
     }
 
-    result.children.emplace_back(std::move(child_projection));
-  }
+    FieldProjection result;
+    result.children.reserve(struct_type.fields().size());
+
+    for (const auto& field : struct_type.fields()) {
+      int32_t field_id = field.field_id();
+      FieldProjection child_projection;
+
+      if (auto iter = field_context_map.find(field_id);
+          iter != field_context_map.cend()) {
+        const auto& parquet_field = iter->second.parquet_field;
+        ICEBERG_ASSIGN_OR_RAISE(child_projection, ProjectField(field, parquet_field,
+                                                               iter->second.local_index));
+      } else if (MetadataColumns::IsMetadataColumn(field_id)) {
+        child_projection.kind = FieldProjection::Kind::kMetadata;
+      } else if (field.initial_default() != nullptr) {
+        // Rows written before the field existed assume its `initial-default` value.
+        child_projection.kind = FieldProjection::Kind::kDefault;
+        child_projection.from = *field.initial_default();
+      } else if (field.optional()) {
+        child_projection.kind = FieldProjection::Kind::kNull;
+      } else {
+        return InvalidSchema("Missing required field with id: {}", field_id);
+      }
 
-  SelectAnchorColumnIfEmpty(&result, parquet_fields);
-  PruneFieldProjection(result);
-  return result;
-}
+      result.children.emplace_back(std::move(child_projection));
+    }
 
-Result<FieldProjection> ProjectList(
-    const ListType& list_type,
-    const std::vector<::parquet::arrow::SchemaField>& parquet_fields) {
-  if (parquet_fields.size() != 1) {
-    return InvalidSchema("List type must have exactly one field, got {}",
-                         parquet_fields.size());
+    SelectAnchorColumnIfEmpty(&result, parquet_fields);
+    PruneFieldProjection(result);
+    return result;
   }
 
-  const auto& parquet_field = parquet_fields.back();
-  auto element_field_id = GetFieldId(parquet_field);
-  if (!element_field_id) {
-    return InvalidSchema("List element field missing field id");
-  }
+  Result<FieldProjection> ProjectList(
+      const ListType& list_type,
+      const std::vector<::parquet::arrow::SchemaField>& parquet_fields) {
+    if (parquet_fields.size() != 1) {
+      return InvalidSchema("List type must have exactly one field, got {}",
+                           parquet_fields.size());
+    }
 
-  const auto& element_field = list_type.fields().back();
-  if (element_field.field_id() != element_field_id.value()) {
-    return InvalidSchema("List element field id mismatch, expected {}, got {}",
-                         element_field.field_id(), element_field_id.value());
-  }
+    const auto& parquet_field = parquet_fields.back();
+    auto element_field_id = GetFieldId(parquet_field);
+    if (!element_field_id) {
+      return InvalidSchema("List element field missing field id");
+    }
 
-  ICEBERG_ASSIGN_OR_RAISE(auto element_projection,
-                          ProjectField(element_field, parquet_field, size_t{0}));
+    const auto& element_field = list_type.fields().back();
+    if (element_field.field_id() != element_field_id.value()) {
+      return InvalidSchema("List element field id mismatch, expected {}, got {}",
+                           element_field.field_id(), element_field_id.value());
+    }
 
-  FieldProjection result;
-  result.children.emplace_back(std::move(element_projection));
-  SelectAnchorColumnIfEmpty(&result, parquet_fields);
-  return result;
-}
+    ICEBERG_ASSIGN_OR_RAISE(auto element_projection,
+                            ProjectField(element_field, parquet_field, size_t{0}));
 
-Result<FieldProjection> ProjectMap(
-    const MapType& map_type,
-    const std::vector<::parquet::arrow::SchemaField>& parquet_fields) {
-  if (parquet_fields.size() != 2) {
-    return InvalidSchema("Map type must have exactly two fields, got {}",
-                         parquet_fields.size());
+    FieldProjection result;
+    result.children.emplace_back(std::move(element_projection));
+    SelectAnchorColumnIfEmpty(&result, parquet_fields);
+    return result;
   }
 
-  auto key_field_id = GetFieldId(parquet_fields[0]);
-  if (!key_field_id) {
-    return InvalidSchema("Map key field missing field id");
-  }
-  auto value_field_id = GetFieldId(parquet_fields[1]);
-  if (!value_field_id) {
-    return InvalidSchema("Map value field missing field id");
-  }
+  Result<FieldProjection> ProjectMap(
+      const MapType& map_type,
+      const std::vector<::parquet::arrow::SchemaField>& parquet_fields) {
+    if (parquet_fields.size() != 2) {
+      return InvalidSchema("Map type must have exactly two fields, got {}",
+                           parquet_fields.size());
+    }
 
-  const auto& key_field = map_type.key();
-  const auto& value_field = map_type.value();
-  if (key_field.field_id() != key_field_id.value()) {
-    return InvalidSchema("Map key field id mismatch, expected {}, got {}",
-                         key_field.field_id(), key_field_id.value());
-  }
-  if (value_field.field_id() != value_field_id.value()) {
-    return InvalidSchema("Map value field id mismatch, expected {}, got {}",
-                         value_field.field_id(), value_field_id.value());
-  }
+    auto key_field_id = GetFieldId(parquet_fields[0]);
+    if (!key_field_id) {
+      return InvalidSchema("Map key field missing field id");
+    }
+    auto value_field_id = GetFieldId(parquet_fields[1]);
+    if (!value_field_id) {
+      return InvalidSchema("Map value field missing field id");
+    }
 
-  FieldProjection result;
-  result.children.reserve(2);
-
-  for (size_t i = 0; i < parquet_fields.size(); ++i) {
-    const auto& sub_node = parquet_fields[i];
-    const auto& sub_field = map_type.fields()[i];
-    ICEBERG_ASSIGN_OR_RAISE(auto sub_projection, ProjectField(sub_field, sub_node, i));
-    if (sub_projection.kind == FieldProjection::Kind::kNull &&
-        !HasSelectedColumn(sub_projection)) {
-      if (auto column_index = FirstColumnIndex(sub_node)) {
-        sub_projection.attributes =
-            std::make_shared<ParquetExtraAttributes>(column_index.value());
-      }
+    const auto& key_field = map_type.key();
+    const auto& value_field = map_type.value();
+    if (key_field.field_id() != key_field_id.value()) {
+      return InvalidSchema("Map key field id mismatch, expected {}, got {}",
+                           key_field.field_id(), key_field_id.value());
+    }
+    if (value_field.field_id() != value_field_id.value()) {
+      return InvalidSchema("Map value field id mismatch, expected {}, got {}",
+                           value_field.field_id(), value_field_id.value());
     }
-    result.children.emplace_back(std::move(sub_projection));
-  }
 
-  SelectAnchorColumnIfEmpty(&result, parquet_fields);
-  return result;
-}
+    FieldProjection result;
+    result.children.reserve(2);
+
+    for (size_t i = 0; i < parquet_fields.size(); ++i) {
+      const auto& sub_node = parquet_fields[i];
+      const auto& sub_field = map_type.fields()[i];
+      ICEBERG_ASSIGN_OR_RAISE(auto sub_projection, ProjectField(sub_field, sub_node, i));
+      if (sub_projection.kind == FieldProjection::Kind::kNull &&
+          !HasSelectedColumn(sub_projection)) {
+        if (auto column_index = FirstColumnIndex(sub_node)) {
+          sub_projection.attributes =
+              std::make_shared<ParquetExtraAttributes>(column_index.value());
+        }
+      }
+      result.children.emplace_back(std::move(sub_projection));
+    }
 
-Result<FieldProjection> ProjectNested(
-    const Type& nested_type,
-    const std::vector<::parquet::arrow::SchemaField>& parquet_fields) {
-  if (!nested_type.is_nested()) {
-    return InvalidSchema("Expected a nested type, but got {}", nested_type);
+    SelectAnchorColumnIfEmpty(&result, parquet_fields);
+    return result;
   }
 
-  switch (nested_type.type_id()) {
-    case TypeId::kStruct:
-      return ProjectStruct(internal::checked_cast<const StructType&>(nested_type),
+  Result<FieldProjection> ProjectNested(
+      const Type& nested_type,
+      const std::vector<::parquet::arrow::SchemaField>& parquet_fields) {
+    if (!nested_type.is_nested()) {
+      return InvalidSchema("Expected a nested type, but got {}", nested_type);
+    }
+
+    switch (nested_type.type_id()) {
+      case TypeId::kStruct:
+        return ProjectStruct(internal::checked_cast<const StructType&>(nested_type),
+                             parquet_fields);
+      case TypeId::kList:
+        return ProjectList(internal::checked_cast<const ListType&>(nested_type),
                            parquet_fields);
-    case TypeId::kList:
-      return ProjectList(internal::checked_cast<const ListType&>(nested_type),
-                         parquet_fields);
-    case TypeId::kMap:
-      if (parquet_fields.size() != 1 ||
-          parquet_fields[0].field->type()->id() != ::arrow::Type::STRUCT ||
-          parquet_fields[0].children.size() != 2) {
-        return InvalidSchema(
-            "Map type must have exactly one struct field with two children");
-      }
-      return ProjectMap(internal::checked_cast<const MapType&>(nested_type),
-                        parquet_fields[0].children);
-    default:
-      return InvalidSchema("Unsupported nested type: {}", nested_type);
+      case TypeId::kMap:
+        if (parquet_fields.size() != 1 ||
+            parquet_fields[0].field->type()->id() != ::arrow::Type::STRUCT ||
+            parquet_fields[0].children.size() != 2) {
+          return InvalidSchema(
+              "Map type must have exactly one struct field with two children");
+        }
+        return ProjectMap(internal::checked_cast<const MapType&>(nested_type),
+                          parquet_fields[0].children);
+      default:
+        return InvalidSchema("Unsupported nested type: {}", nested_type);
+    }
   }
-}
+
+  const ::parquet::SchemaDescriptor* descr_;
+};
 
 void CollectColumnIds(const FieldProjection& field_projection,
                       std::vector<int32_t>* column_ids) {
@@ -484,9 +717,10 @@ void CollectColumnIds(const FieldProjection& field_projection,
 
 Result<SchemaProjection> Project(const Schema& expected_schema,
                                  const ::parquet::arrow::SchemaManifest& parquet_schema) {
+  ProjectionBuilder builder(parquet_schema.descr);
   ICEBERG_ASSIGN_OR_RAISE(auto field_projection,
-                          ProjectNested(static_cast<const Type&>(expected_schema),
-                                        parquet_schema.schema_fields));
+                          builder.Project(static_cast<const Type&>(expected_schema),
+                                          parquet_schema.schema_fields));
   return SchemaProjection{std::move(field_projection.children)};
 }
 
@@ -516,4 +750,20 @@ bool HasFieldIds(const ::parquet::schema::NodePtr& node) {
   return false;
 }
 
+Result<std::shared_ptr<::parquet::SchemaDescriptor>> ToParquetSchema(
+    const Schema& iceberg_schema) {
+  ::parquet::schema::NodeVector fields;
+  fields.reserve(iceberg_schema.fields().size());
+  for (const auto& field : iceberg_schema.fields()) {
+    ICEBERG_ASSIGN_OR_RAISE(auto parquet_field, ToParquetNode(field));
+    fields.push_back(std::move(parquet_field));
+  }
+
+  auto root = ::parquet::schema::GroupNode::Make(
+      "schema", ::parquet::Repetition::REQUIRED, std::move(fields));
+  auto parquet_schema = std::make_shared<::parquet::SchemaDescriptor>();
+  parquet_schema->Init(std::move(root));
+  return parquet_schema;
+}
+
 }  // namespace iceberg::parquet
diff --git a/src/iceberg/parquet/parquet_schema_util_internal.h b/src/iceberg/parquet/parquet_schema_util_internal.h
index 567069291..2eafd013b 100644
--- a/src/iceberg/parquet/parquet_schema_util_internal.h
+++ b/src/iceberg/parquet/parquet_schema_util_internal.h
@@ -22,6 +22,7 @@
 #include <optional>
 
 #include <parquet/arrow/schema.h>
+#include <parquet/type_fwd.h>
 
 #include "iceberg/schema.h"
 #include "iceberg/schema_util.h"
@@ -62,8 +63,8 @@ std::vector<int32_t> SelectedColumnIndices(const SchemaProjection& projection);
 /// \return True if the Parquet schema has field IDs, false otherwise.
 bool HasFieldIds(const ::parquet::schema::NodePtr& root_node);
 
-/// \brief Validate whether a projected Iceberg type is compatible with a Parquet field.
-Status ValidateParquetSchemaEvolution(const Type& expected_type,
-                                      const ::parquet::arrow::SchemaField& parquet_field);
+/// \brief Convert an Iceberg schema to a Parquet schema.
+Result<std::shared_ptr<::parquet::SchemaDescriptor>> ToParquetSchema(
+    const Schema& iceberg_schema);
 
 }  // namespace iceberg::parquet
diff --git a/src/iceberg/parquet/parquet_writer.cc b/src/iceberg/parquet/parquet_writer.cc
index fea7cd834..08c394138 100644
--- a/src/iceberg/parquet/parquet_writer.cc
+++ b/src/iceberg/parquet/parquet_writer.cc
@@ -33,7 +33,6 @@
 #include <arrow/record_batch.h>
 #include <arrow/util/compression.h>
 #include <arrow/util/key_value_metadata.h>
-#include <parquet/arrow/schema.h>
 #include <parquet/arrow/writer.h>
 #include <parquet/file_writer.h>
 #include <parquet/properties.h>
@@ -41,6 +40,7 @@
 #include "iceberg/arrow/arrow_io_internal.h"
 #include "iceberg/arrow/arrow_status_internal.h"
 #include "iceberg/parquet/parquet_metrics_internal.h"
+#include "iceberg/parquet/parquet_schema_util_internal.h"
 #include "iceberg/schema_internal.h"
 #include "iceberg/type.h"
 #include "iceberg/util/macros.h"
@@ -260,9 +260,7 @@ class ParquetWriter::Impl {
     ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(*schema_, &c_schema));
     ICEBERG_ARROW_ASSIGN_OR_RETURN(arrow_schema_, ::arrow::ImportSchema(&c_schema));
 
-    ICEBERG_ARROW_RETURN_NOT_OK(
-        ::parquet::arrow::ToParquetSchema(arrow_schema_.get(), *writer_properties,
-                                          *arrow_writer_properties, &parquet_schema_));
+    ICEBERG_ASSIGN_OR_RAISE(parquet_schema_, ToParquetSchema(*schema_));
     auto schema_node = std::static_pointer_cast<::parquet::schema::GroupNode>(
         parquet_schema_->schema_root());
 
diff --git a/src/iceberg/schema_internal.cc b/src/iceberg/schema_internal.cc
index 810e53695..baa8b1b2f 100644
--- a/src/iceberg/schema_internal.cc
+++ b/src/iceberg/schema_internal.cc
@@ -43,8 +43,6 @@ constexpr int32_t kUnknownFieldId = -1;
 Status CheckArrowCompatible(const Type& type) {
   switch (type.type_id()) {
     case TypeId::kVariant:
-    case TypeId::kGeometry:
-    case TypeId::kGeography:
       return NotSupported("Iceberg type {} is not supported by Arrow conversion",
                           type.ToString());
     case TypeId::kStruct:
@@ -165,6 +163,8 @@ ArrowErrorCode ToArrowSchema(const Type& type, bool optional, std::string_view n
       NANOARROW_RETURN_NOT_OK(ArrowSchemaSetType(schema, NANOARROW_TYPE_STRING));
       break;
     case TypeId::kBinary:
+    case TypeId::kGeometry:
+    case TypeId::kGeography:
       NANOARROW_RETURN_NOT_OK(ArrowSchemaSetType(schema, NANOARROW_TYPE_BINARY));
       break;
     case TypeId::kFixed: {
@@ -183,8 +183,7 @@ ArrowErrorCode ToArrowSchema(const Type& type, bool optional, std::string_view n
       NANOARROW_RETURN_NOT_OK(ArrowSchemaSetType(schema, NANOARROW_TYPE_NA));
       break;
     case TypeId::kVariant:
-    case TypeId::kGeometry:
-    case TypeId::kGeography:
+      ArrowBufferReset(&metadata_buffer);
       return EINVAL;
   }
 
diff --git a/src/iceberg/test/arrow_test.cc b/src/iceberg/test/arrow_test.cc
index 1ba95cdd1..5720b42fe 100644
--- a/src/iceberg/test/arrow_test.cc
+++ b/src/iceberg/test/arrow_test.cc
@@ -20,6 +20,8 @@
 #include <functional>
 #include <memory>
 #include <mutex>
+#include <string>
+#include <string_view>
 #include <thread>
 #include <unordered_map>
 #include <utility>
@@ -115,6 +117,10 @@ INSTANTIATE_TEST_SUITE_P(
                            .arrow_type = ::arrow::utf8()},
         ToArrowSchemaParam{.iceberg_type = iceberg::binary(),
                            .arrow_type = ::arrow::binary()},
+        ToArrowSchemaParam{.iceberg_type = iceberg::geometry(),
+                           .arrow_type = ::arrow::binary()},
+        ToArrowSchemaParam{.iceberg_type = iceberg::geography(),
+                           .arrow_type = ::arrow::binary()},
         ToArrowSchemaParam{.iceberg_type = iceberg::uuid(),
                            .arrow_type = ::arrow::extension::uuid()},
         ToArrowSchemaParam{.iceberg_type = iceberg::fixed(20),
@@ -123,8 +129,7 @@ INSTANTIATE_TEST_SUITE_P(
                            .arrow_type = ::arrow::null()}));
 
 TEST(ToArrowSchemaTest, UnsupportedV3Types) {
-  const std::vector<std::shared_ptr<Type>> unsupported_types = {
-      iceberg::variant(), iceberg::geometry(), iceberg::geography()};
+  const std::vector<std::shared_ptr<Type>> unsupported_types = {iceberg::variant()};
 
   for (const auto& unsupported_type : unsupported_types) {
     Schema schema(
diff --git a/src/iceberg/test/avro_schema_test.cc b/src/iceberg/test/avro_schema_test.cc
index 169eff64c..a65a0abdd 100644
--- a/src/iceberg/test/avro_schema_test.cc
+++ b/src/iceberg/test/avro_schema_test.cc
@@ -245,9 +245,14 @@ TEST(ToAvroNodeVisitorTest, FixedType) {
 }
 
 TEST(ToAvroNodeVisitorTest, BinaryType) {
-  ::avro::NodePtr node;
-  EXPECT_THAT(ToAvroNodeVisitor{}.Visit(BinaryType{}, &node), IsOk());
-  EXPECT_EQ(node->type(), ::avro::AVRO_BYTES);
+  auto assert_bytes = [](const auto& type) {
+    ::avro::NodePtr node;
+    EXPECT_THAT(ToAvroNodeVisitor{}.Visit(type, &node), IsOk());
+    EXPECT_EQ(node->type(), ::avro::AVRO_BYTES);
+  };
+  assert_bytes(BinaryType{});
+  assert_bytes(*iceberg::geometry());
+  assert_bytes(*iceberg::geography());
 }
 
 TEST(ToAvroNodeVisitorTest, UnknownType) {
@@ -777,6 +782,32 @@ TEST(AvroSchemaProjectionTest, ProjectIdenticalSchemas) {
   }
 }
 
+TEST(AvroSchemaProjectionTest, ProjectGeospatialTypesFromBytes) {
+  Schema expected_schema({
+      SchemaField::MakeRequired(/*field_id=*/1, "geom", iceberg::geometry()),
+      SchemaField::MakeOptional(/*field_id=*/2, "geog", iceberg::geography()),
+  });
+  auto avro_schema = ::avro::compileJsonSchemaFromString(R"({
+    "type": "record",
+    "name": "iceberg_schema",
+    "fields": [
+      {"name": "geog", "type": ["null", "bytes"], "field-id": 2},
+      {"name": "geom", "type": "bytes", "field-id": 1}
+    ]
+  })");
+
+  auto projection_result =
+      Project(expected_schema, avro_schema.root(), /*prune_source=*/false);
+  ASSERT_THAT(projection_result, IsOk());
+
+  const auto& projection = *projection_result;
+  ASSERT_EQ(projection.fields.size(), 2);
+  EXPECT_EQ(projection.fields[0].kind, FieldProjection::Kind::kProjected);
+  EXPECT_EQ(std::get<1>(projection.fields[0].from), 1);
+  EXPECT_EQ(projection.fields[1].kind, FieldProjection::Kind::kProjected);
+  EXPECT_EQ(std::get<1>(projection.fields[1].from), 0);
+}
+
 TEST(AvroSchemaProjectionTest, ProjectSubsetSchema) {
   // Create a subset iceberg schema
   Schema expected_schema({
diff --git a/src/iceberg/test/avro_test.cc b/src/iceberg/test/avro_test.cc
index 20156a57c..d322c3a8f 100644
--- a/src/iceberg/test/avro_test.cc
+++ b/src/iceberg/test/avro_test.cc
@@ -50,6 +50,7 @@
 #include "iceberg/test/matchers.h"
 #include "iceberg/test/std_io.h"
 #include "iceberg/test/temp_file_test_base.h"
+#include "iceberg/test/test_resource.h"
 #include "iceberg/type.h"
 #include "iceberg/util/checked_cast.h"
 #include "iceberg/util/uuid.h"
@@ -1079,6 +1080,37 @@ TEST_P(AvroWriterTest, WriteUuidType) {
   ASSERT_NO_FATAL_FAILURE(VerifyExhausted(*reader));
 }
 
+TEST_P(AvroWriterTest, WriteGeospatialTypesAsOpaqueBytes) {
+  auto schema = std::make_shared<iceberg::Schema>(std::vector<SchemaField>{
+      SchemaField::MakeOptional(1, "geom", iceberg::geometry()),
+      SchemaField::MakeRequired(
+          2, "nested",
+          iceberg::struct_({SchemaField::MakeOptional(
+              3, "geog", iceberg::geography("EPSG:4326", EdgeAlgorithm::kVincenty))})),
+  });
+
+  ::arrow::BinaryBuilder geometry_builder;
+  const std::array<uint8_t, 3> geometry_bytes = {0xff, 0x00, 0x42};
+  ASSERT_TRUE(geometry_builder.Append(geometry_bytes.data(), geometry_bytes.size()).ok());
+  ASSERT_TRUE(geometry_builder.AppendNull().ok());
+
+  ::arrow::BinaryBuilder geography_builder;
+  const std::array<uint8_t, 4> geography_bytes = {0x01, 0x02, 0x03, 0x04};
+  ASSERT_TRUE(geography_builder.AppendNull().ok());
+  ASSERT_TRUE(
+      geography_builder.Append(geography_bytes.data(), geography_bytes.size()).ok());
+  auto nested = ::arrow::StructArray::Make({geography_builder.Finish().ValueOrDie()},
+                                           {::arrow::field("geog", ::arrow::binary())})
+                    .ValueOrDie();
+
+  auto array = ::arrow::StructArray::Make(
+                   {geometry_builder.Finish().ValueOrDie(), nested},
+                   {::arrow::field("geom", ::arrow::binary()),
+                    ::arrow::field("nested", nested->type(), /*nullable=*/false)})
+                   .ValueOrDie();
+  ASSERT_NO_FATAL_FAILURE(WriteArrowArrayAndVerify(schema, array));
+}
+
 TEST_P(AvroWriterTest, WriteUuidListType) {
   auto schema = std::make_shared<iceberg::Schema>(std::vector<SchemaField>{
       SchemaField::MakeRequired(1, "uuid_list",
diff --git a/src/iceberg/test/parquet_metrics_test.cc b/src/iceberg/test/parquet_metrics_test.cc
index 6afe6f9c8..e5893e180 100644
--- a/src/iceberg/test/parquet_metrics_test.cc
+++ b/src/iceberg/test/parquet_metrics_test.cc
@@ -177,4 +177,21 @@ TEST_F(ParquetMetricsTest, UnrepresentableTruncatedUpperBoundIsOmitted) {
                                      std::nullopt, metrics);
 }
 
+TEST_F(ParquetMetricsTest, GeospatialMetricsExcludeBounds) {
+  auto schema = std::make_shared<Schema>(std::vector<SchemaField>{
+      SchemaField::MakeOptional(1, "geom", geometry()),
+  });
+  ::arrow::BinaryBuilder builder;
+  const std::vector<uint8_t> wkb = {0xff, 0x00, 0x42};
+  ASSERT_TRUE(builder.Append(wkb.data(), wkb.size()).ok());
+  ASSERT_TRUE(builder.AppendNull().ok());
+  auto records =
+      ::arrow::StructArray::Make({builder.Finish().ValueOrDie()},
+                                 {::arrow::field("geom", ::arrow::binary(), true)})
+          .ValueOrDie();
+
+  ICEBERG_UNWRAP_OR_FAIL(auto metrics, GetMetrics(schema, records));
+  AssertBounds<std::vector<uint8_t>>(1, geometry(), std::nullopt, std::nullopt, metrics);
+}
+
 }  // namespace iceberg::test
diff --git a/src/iceberg/test/parquet_schema_test.cc b/src/iceberg/test/parquet_schema_test.cc
index 75e99ff12..41787d806 100644
--- a/src/iceberg/test/parquet_schema_test.cc
+++ b/src/iceberg/test/parquet_schema_test.cc
@@ -17,8 +17,10 @@
  * under the License.
  */
 
+#include <array>
 #include <string>
 #include <string_view>
+#include <utility>
 #include <vector>
 
 #include <arrow/type.h>
@@ -33,6 +35,7 @@
 #include "iceberg/schema.h"
 #include "iceberg/test/matchers.h"
 #include "iceberg/type.h"
+#include "iceberg/util/checked_cast.h"
 
 namespace iceberg::parquet {
 
@@ -112,8 +115,10 @@ ::parquet::schema::NodePtr MakeMapNode(const std::string& name,
 // Helper to create SchemaManifest from Parquet schema
 ::parquet::arrow::SchemaManifest MakeSchemaManifest(
     const ::parquet::schema::NodePtr& parquet_schema) {
+  static std::vector<std::shared_ptr<::parquet::SchemaDescriptor>> descriptors;
   auto parquet_schema_descriptor = std::make_shared<::parquet::SchemaDescriptor>();
   parquet_schema_descriptor->Init(parquet_schema);
+  descriptors.push_back(parquet_schema_descriptor);
 
   auto properties = ::parquet::default_arrow_reader_properties();
   properties.set_arrow_extensions_enabled(true);
@@ -128,6 +133,15 @@ ::parquet::arrow::SchemaManifest MakeSchemaManifest(
   return manifest;
 }
 
+::parquet::schema::NodePtr MakeBinaryNode(
+    const std::string& name, std::shared_ptr<const ::parquet::LogicalType> logical_type,
+    int field_id) {
+  return ::parquet::schema::PrimitiveNode::Make(name, ::parquet::Repetition::OPTIONAL,
+                                                std::move(logical_type),
+                                                ::parquet::Type::BYTE_ARRAY,
+                                                /*primitive_length=*/-1, field_id);
+}
+
 ::parquet::arrow::SchemaField MakeNullSchemaField(const std::string& name, int field_id) {
   ::parquet::arrow::SchemaField schema_field;
   schema_field.field =
@@ -332,12 +346,107 @@ TEST(ParquetSchemaProjectionTest, ProjectSchemaEvolutionFloatToDouble) {
   ASSERT_PROJECTED_FIELD(projection.fields[0], 0);
 }
 
-TEST(ParquetSchemaProjectionTest, ValidateSchemaEvolutionAllowsNullPhysicalType) {
-  ::parquet::arrow::SchemaField parquet_field;
-  parquet_field.field = ::arrow::field("value", ::arrow::null());
+TEST(ParquetSchemaConversionTest, DecimalPhysicalTypes) {
+  Schema schema({
+      SchemaField::MakeRequired(/*field_id=*/1, "p9", decimal(9, 2)),
+      SchemaField::MakeRequired(/*field_id=*/2, "p10", decimal(10, 2)),
+      SchemaField::MakeRequired(/*field_id=*/3, "p18", decimal(18, 2)),
+      SchemaField::MakeRequired(/*field_id=*/4, "p19", decimal(19, 2)),
+  });
+
+  ICEBERG_UNWRAP_OR_FAIL(auto parquet_schema, ToParquetSchema(schema));
+  EXPECT_EQ(parquet_schema->Column(0)->physical_type(), ::parquet::Type::INT32);
+  EXPECT_EQ(parquet_schema->Column(1)->physical_type(), ::parquet::Type::INT64);
+  EXPECT_EQ(parquet_schema->Column(2)->physical_type(), ::parquet::Type::INT64);
+  EXPECT_EQ(parquet_schema->Column(3)->physical_type(),
+            ::parquet::Type::FIXED_LEN_BYTE_ARRAY);
+  EXPECT_EQ(parquet_schema->Column(3)->type_length(), 9);
+}
+
+TEST(ParquetGeospatialSchemaTest, ConvertsGeospatialTypes) {
+  using ParquetAlgorithm = ::parquet::LogicalType::EdgeInterpolationAlgorithm;
+  const std::array<std::pair<EdgeAlgorithm, ParquetAlgorithm>, 5> algorithms = {{
+      {EdgeAlgorithm::kSpherical, ParquetAlgorithm::SPHERICAL},
+      {EdgeAlgorithm::kVincenty, ParquetAlgorithm::VINCENTY},
+      {EdgeAlgorithm::kThomas, ParquetAlgorithm::THOMAS},
+      {EdgeAlgorithm::kAndoyer, ParquetAlgorithm::ANDOYER},
+      {EdgeAlgorithm::kKarney, ParquetAlgorithm::KARNEY},
+  }};
+  std::vector<SchemaField> fields = {
+      SchemaField::MakeOptional(
+          /*field_id=*/1, "struct_field",
+          std::make_shared<StructType>(std::vector<SchemaField>{SchemaField::MakeOptional(
+              /*field_id=*/2, "geom", iceberg::geometry("EPSG:3857"))})),
+      SchemaField::MakeOptional(
+          /*field_id=*/3, "list_field",
+          std::make_shared<ListType>(SchemaField::MakeOptional(
+              /*field_id=*/4, "element", iceberg::geometry()))),
+      SchemaField::MakeOptional(
+          /*field_id=*/5, "map_field",
+          std::make_shared<MapType>(
+              SchemaField::MakeRequired(/*field_id=*/6, "key", iceberg::string()),
+              SchemaField::MakeOptional(/*field_id=*/7, "value", iceberg::geometry()))),
+  };
+  for (size_t index = 0; index < algorithms.size(); ++index) {
+    fields.push_back(SchemaField::MakeOptional(
+        static_cast<int32_t>(index + 8), "geog_" + std::to_string(index),
+        iceberg::geography("EPSG:4326", algorithms[index].first)));
+  }
+  Schema schema(std::move(fields));
+
+  ICEBERG_UNWRAP_OR_FAIL(auto parquet_schema, ToParquetSchema(schema));
+  ASSERT_EQ(parquet_schema->num_columns(), 9);
+  const auto& geometry_logical =
+      internal::checked_cast<const ::parquet::GeometryLogicalType&>(
+          *parquet_schema->Column(0)->logical_type());
+  ASSERT_EQ(geometry_logical.crs(), "EPSG:3857");
+  ASSERT_TRUE(parquet_schema->Column(1)->logical_type()->is_geometry());
+  ASSERT_TRUE(parquet_schema->Column(3)->logical_type()->is_geometry());
+
+  for (size_t index = 0; index < algorithms.size(); ++index) {
+    const auto& geography_logical =
+        internal::checked_cast<const ::parquet::GeographyLogicalType&>(
+            *parquet_schema->Column(static_cast<int>(index + 4))->logical_type());
+    ASSERT_EQ(geography_logical.crs(), "EPSG:4326");
+    ASSERT_EQ(geography_logical.algorithm(), algorithms[index].second);
+  }
+}
+
+TEST(ParquetGeospatialSchemaTest, ProjectsTypesWithDifferentParameters) {
+  Schema expected_schema({
+      SchemaField::MakeOptional(/*field_id=*/1, "geom", iceberg::geometry()),
+      SchemaField::MakeOptional(
+          /*field_id=*/2, "geog",
+          iceberg::geography("epsg:4326", EdgeAlgorithm::kKarney)),
+  });
+  auto parquet_schema = MakeGroupNode(
+      "iceberg_schema",
+      {MakeBinaryNode("geom", ::parquet::LogicalType::Geometry("EPSG:3857"),
+                      /*field_id=*/1),
+       MakeBinaryNode("geog",
+                      ::parquet::LogicalType::Geography(
+                          "OGC:CRS84",
+                          ::parquet::LogicalType::EdgeInterpolationAlgorithm::SPHERICAL),
+                      /*field_id=*/2)});
+  auto manifest = MakeSchemaManifest(parquet_schema);
+  ICEBERG_UNWRAP_OR_FAIL(auto projection, Project(expected_schema, manifest));
+  ASSERT_EQ(projection.fields.size(), 2);
+  ASSERT_PROJECTED_FIELD(projection.fields[0], 0);
+  ASSERT_PROJECTED_FIELD(projection.fields[1], 1);
+}
 
-  auto status = ValidateParquetSchemaEvolution(*iceberg::int32(), parquet_field);
-  ASSERT_THAT(status, IsOk());
+TEST(ParquetGeospatialSchemaTest, RejectsMissingLogicalType) {
+  Schema geometry_schema({
+      SchemaField::MakeOptional(/*field_id=*/1, "geom", iceberg::geometry()),
+  });
+  auto missing_logical_type = MakeGroupNode(
+      "iceberg_schema",
+      {MakeBinaryNode("geom", ::parquet::LogicalType::None(), /*field_id=*/1)});
+  auto missing_logical_type_manifest = MakeSchemaManifest(missing_logical_type);
+  auto result = Project(geometry_schema, missing_logical_type_manifest);
+  ASSERT_THAT(result, IsError(ErrorKind::kInvalidSchema));
+  ASSERT_THAT(result,
+              HasErrorMessage("Iceberg geometry requires Parquet Geometry logical type"));
 }
 
 TEST(ParquetSchemaProjectionTest, ProjectNullPhysicalFieldsAsNull) {
diff --git a/src/iceberg/test/parquet_test.cc b/src/iceberg/test/parquet_test.cc
index 6ccff6cb4..d2cdad8f7 100644
--- a/src/iceberg/test/parquet_test.cc
+++ b/src/iceberg/test/parquet_test.cc
@@ -53,6 +53,7 @@
 #include "iceberg/test/matchers.h"
 #include "iceberg/test/std_io.h"
 #include "iceberg/test/temp_file_test_base.h"
+#include "iceberg/test/test_resource.h"
 #include "iceberg/type.h"
 #include "iceberg/util/checked_cast.h"
 #include "iceberg/util/macros.h"
@@ -953,4 +954,55 @@ TEST_F(ParquetReadWrite, UuidRoundTrip) {
                                    << array->ToString();
 }
 
+TEST_F(ParquetReadWrite, GeospatialWkbRoundTrip) {
+  auto schema = std::make_shared<Schema>(std::vector<SchemaField>{
+      SchemaField::MakeOptional(1, "geom", geometry()),
+      SchemaField::MakeOptional(2, "geog",
+                                geography("EPSG:4326", EdgeAlgorithm::kAndoyer)),
+      SchemaField::MakeOptional(
+          3, "nested",
+          struct_({SchemaField::MakeOptional(4, "geom", geometry("EPSG:3857"))})),
+  });
+
+  ::arrow::BinaryBuilder geometry_builder;
+  const std::array<uint8_t, 4> arbitrary_geometry = {0xff, 0x00, 0x7f, 0x42};
+  ASSERT_TRUE(
+      geometry_builder.Append(arbitrary_geometry.data(), arbitrary_geometry.size()).ok());
+  ASSERT_TRUE(geometry_builder.AppendNull().ok());
+
+  ::arrow::BinaryBuilder geography_builder;
+  const std::array<uint8_t, 5> arbitrary_geography = {0x01, 0x02, 0x03, 0x04, 0x05};
+  ASSERT_TRUE(geography_builder.AppendNull().ok());
+  ASSERT_TRUE(
+      geography_builder.Append(arbitrary_geography.data(), arbitrary_geography.size())
+          .ok());
+
+  ::arrow::BinaryBuilder nested_geometry_builder;
+  const std::array<uint8_t, 3> arbitrary_nested_geometry = {0xde, 0xad, 0xbe};
+  ASSERT_TRUE(
+      nested_geometry_builder
+          .Append(arbitrary_nested_geometry.data(), arbitrary_nested_geometry.size())
+          .ok());
+  ASSERT_TRUE(nested_geometry_builder.AppendNull().ok());
+  auto nested =
+      ::arrow::StructArray::Make({nested_geometry_builder.Finish().ValueOrDie()},
+                                 {::arrow::field("geom", ::arrow::binary())})
+          .ValueOrDie();
+
+  auto array =
+      ::arrow::StructArray::Make({geometry_builder.Finish().ValueOrDie(),
+                                  geography_builder.Finish().ValueOrDie(), nested},
+                                 {::arrow::field("geom", ::arrow::binary()),
+                                  ::arrow::field("geog", ::arrow::binary()),
+                                  ::arrow::field("nested", nested->type())})
+          .ValueOrDie();
+
+  std::shared_ptr<::arrow::Array> out;
+  DoRoundtrip(array, schema, out);
+  ASSERT_NE(out, nullptr);
+  ASSERT_TRUE(out->Equals(*array)) << "actual:\n"
+                                   << out->ToString() << "\nexpected:\n"
+                                   << array->ToString();
+}
+
 }  // namespace iceberg::parquet
diff --git a/src/iceberg/test/schema_json_test.cc b/src/iceberg/test/schema_json_test.cc
index c946550cc..f90a0c2ca 100644
--- a/src/iceberg/test/schema_json_test.cc
+++ b/src/iceberg/test/schema_json_test.cc
@@ -67,18 +67,13 @@ INSTANTIATE_TEST_SUITE_P(
         SchemaJsonParam{.json = "\"uuid\"", .type = iceberg::uuid()},
         SchemaJsonParam{.json = "\"unknown\"", .type = iceberg::unknown()},
         SchemaJsonParam{.json = "\"variant\"", .type = iceberg::variant()},
-        SchemaJsonParam{.json = "\"geometry\"", .type = iceberg::geometry()},
+        SchemaJsonParam{.json = "\"geometry(OGC:CRS84)\"", .type = iceberg::geometry()},
         SchemaJsonParam{.json = "\"geometry(srid:4326)\"",
                         .type = iceberg::geometry("srid:4326")},
-        SchemaJsonParam{.json = "\"geography\"", .type = iceberg::geography()},
-        SchemaJsonParam{.json = "\"geography(srid:4326)\"",
+        SchemaJsonParam{.json = "\"geography(OGC:CRS84, spherical)\"",
+                        .type = iceberg::geography()},
+        SchemaJsonParam{.json = "\"geography(srid:4326, spherical)\"",
                         .type = iceberg::geography("srid:4326")},
-        SchemaJsonParam{
-            .json = "\"geography(srid:4326, spherical)\"",
-            .type = iceberg::geography("srid:4326", EdgeAlgorithm::kSpherical)},
-        SchemaJsonParam{
-            .json = "\"geography(OGC:CRS84, spherical)\"",
-            .type = iceberg::geography("OGC:CRS84", EdgeAlgorithm::kSpherical)},
         SchemaJsonParam{.json = "\"geography(srid:4326, karney)\"",
                         .type = iceberg::geography("srid:4326", EdgeAlgorithm::kKarney)},
         SchemaJsonParam{.json = "\"fixed[8]\"", .type = iceberg::fixed(8)},
@@ -144,14 +139,20 @@ TEST(TypeJsonTest, FromJsonV3TypesWithSpacesAndCase) {
             *iceberg::geography("srid:4269", EdgeAlgorithm::kKarney));
 }
 
+TEST(TypeJsonTest, FromJsonAcceptsBareGeospatialTypes) {
+  ICEBERG_UNWRAP_OR_FAIL(auto geometry_type,
+                         TypeFromJson(nlohmann::json::parse("\"geometry\"")));
+  ASSERT_EQ(*geometry_type, *iceberg::geometry());
+
+  ICEBERG_UNWRAP_OR_FAIL(auto geography_type,
+                         TypeFromJson(nlohmann::json::parse("\"geography\"")));
+  ASSERT_EQ(*geography_type, *iceberg::geography());
+}
+
 TEST(TypeJsonTest, InvalidV3Types) {
   auto invalid_geometry = TypeFromJson(nlohmann::json::parse("\"geometry()\""));
   ASSERT_THAT(invalid_geometry, HasErrorMessage("Invalid geometry type"));
 
-  auto invalid_geometry_with_spaces =
-      TypeFromJson(nlohmann::json::parse("\"geometry( )\""));
-  ASSERT_THAT(invalid_geometry_with_spaces, HasErrorMessage("Invalid geometry type"));
-
   auto invalid_geography = TypeFromJson(nlohmann::json::parse("\"geography()\""));
   ASSERT_THAT(invalid_geography, HasErrorMessage("Invalid geography type"));
 
diff --git a/src/iceberg/test/type_test.cc b/src/iceberg/test/type_test.cc
index ac188f229..b3a6ea622 100644
--- a/src/iceberg/test/type_test.cc
+++ b/src/iceberg/test/type_test.cc
@@ -19,15 +19,18 @@
 
 #include "iceberg/type.h"
 
+#include <cstdint>
 #include <format>
 #include <memory>
 #include <string>
 #include <thread>
+#include <vector>
 
 #include <gmock/gmock.h>
 #include <gtest/gtest.h>
 
 #include "iceberg/exception.h"
+#include "iceberg/geospatial.h"
 #include "iceberg/test/matchers.h"
 #include "iceberg/util/formatter.h"  // IWYU pragma: keep
 #include "iceberg/util/type_util.h"
@@ -241,14 +244,14 @@ const static std::array<TypeTestCase, 21> kPrimitiveTypes = {{
         .type = iceberg::geometry(),
         .type_id = iceberg::TypeId::kGeometry,
         .primitive = true,
-        .repr = "geometry",
+        .repr = "geometry(OGC:CRS84)",
     },
     {
         .name = "geography",
         .type = iceberg::geography(),
         .type_id = iceberg::TypeId::kGeography,
         .primitive = true,
-        .repr = "geography",
+        .repr = "geography(OGC:CRS84, spherical)",
     },
 }};
 
@@ -336,31 +339,173 @@ TEST(TypeTest, Equality) {
   }
 }
 
-TEST(TypeTest, GeographyExplicitDefaultAlgorithm) {
-  ASSERT_NE(*iceberg::geography("srid:4326"),
-            *iceberg::geography("srid:4326", iceberg::EdgeAlgorithm::kSpherical));
-  ASSERT_NE(*iceberg::geography(),
+TEST(TypeTest, GeospatialTypes) {
+  ASSERT_EQ(*iceberg::geometry(), *iceberg::geometry("ogc:crs84"));
+  ASSERT_EQ(*iceberg::geometry("EPSG:4326"), *iceberg::geometry("epsg:4326"));
+  ASSERT_EQ("geometry(epsg:4326)", iceberg::geometry("epsg:4326")->ToString());
+
+  ASSERT_EQ(*iceberg::geography(),
             *iceberg::geography("OGC:CRS84", iceberg::EdgeAlgorithm::kSpherical));
-  ASSERT_EQ(
-      "geography(srid:4326, spherical)",
-      iceberg::geography("srid:4326", iceberg::EdgeAlgorithm::kSpherical)->ToString());
-  ASSERT_EQ(
-      "geography(OGC:CRS84, spherical)",
-      iceberg::geography("OGC:CRS84", iceberg::EdgeAlgorithm::kSpherical)->ToString());
+  ASSERT_EQ(*iceberg::geography("srid:4326"),
+            *iceberg::geography("SRID:4326", iceberg::EdgeAlgorithm::kSpherical));
+  ASSERT_EQ("geography(srid:4326, spherical)",
+            iceberg::geography("srid:4326")->ToString());
   ASSERT_NE(*iceberg::geography("srid:4326"),
             *iceberg::geography("srid:4326", iceberg::EdgeAlgorithm::kKarney));
+
+  auto geometry_result = iceberg::GeometryType::Make("");
+  ASSERT_THAT(geometry_result, IsError(iceberg::ErrorKind::kInvalidArgument));
+  ASSERT_THAT(geometry_result,
+              iceberg::HasErrorMessage("GeometryType: CRS cannot be empty"));
+
+  auto geography_result = iceberg::GeographyType::Make("");
+  ASSERT_THAT(geography_result, IsError(iceberg::ErrorKind::kInvalidArgument));
+  ASSERT_THAT(geography_result,
+              iceberg::HasErrorMessage("GeographyType: CRS cannot be empty"));
 }
 
-TEST(TypeTest, GeometryMakeRejectsEmptyCrs) {
-  auto result = iceberg::GeometryType::Make("");
-  ASSERT_THAT(result, IsError(iceberg::ErrorKind::kInvalidArgument));
-  ASSERT_THAT(result, iceberg::HasErrorMessage("GeometryType: CRS cannot be empty"));
+TEST(GeospatialBoundTest, Encoding) {
+  const auto xy = iceberg::GeospatialBound::XY(1.0, -2.0);
+  const std::vector<uint8_t> xy_bytes = {
+      0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xf0, 0x3f,
+      0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xc0,
+  };
+  ASSERT_EQ(xy.Serialize(), xy_bytes);
+  ICEBERG_UNWRAP_OR_FAIL(auto deserialized_xy,
+                         iceberg::GeospatialBound::Deserialize(xy_bytes));
+  ASSERT_EQ(deserialized_xy, xy);
+
+  for (const auto& expected : {iceberg::GeospatialBound::XYZ(1.0, 2.0, 3.0),
+                               iceberg::GeospatialBound::XYM(1.0, 2.0, 4.0),
+                               iceberg::GeospatialBound::XYZM(1.0, 2.0, 3.0, 4.0)}) {
+    ICEBERG_UNWRAP_OR_FAIL(auto actual,
+                           iceberg::GeospatialBound::Deserialize(expected.Serialize()));
+    ASSERT_EQ(actual, expected);
+  }
+
+  ASSERT_THAT(iceberg::GeospatialBound::Deserialize(std::vector<uint8_t>(15)),
+              IsError(iceberg::ErrorKind::kInvalidArgument));
 }
 
-TEST(TypeTest, GeographyMakeRejectsEmptyCrs) {
-  auto result = iceberg::GeographyType::Make("");
-  ASSERT_THAT(result, IsError(iceberg::ErrorKind::kInvalidArgument));
-  ASSERT_THAT(result, iceberg::HasErrorMessage("GeographyType: CRS cannot be empty"));
+TEST(BoundingBoxTest, Encoding) {
+  iceberg::BoundingBox expected(iceberg::GeospatialBound::XY(0.0, 1.0),
+                                iceberg::GeospatialBound::XYZ(2.0, 3.0, 4.0));
+  ICEBERG_UNWRAP_OR_FAIL(auto from_separate,
+                         iceberg::BoundingBox::Deserialize(expected.lower().Serialize(),
+                                                           expected.upper().Serialize()));
+  ASSERT_EQ(from_separate, expected);
+
+  auto combined = expected.Serialize();
+  ASSERT_EQ(combined.size(), 48);
+  ASSERT_EQ(std::vector<uint8_t>(combined.begin(), combined.begin() + 4),
+            (std::vector<uint8_t>{0x10, 0x00, 0x00, 0x00}));
+  ASSERT_EQ(std::vector<uint8_t>(combined.begin() + 20, combined.begin() + 24),
+            (std::vector<uint8_t>{0x18, 0x00, 0x00, 0x00}));
+  ICEBERG_UNWRAP_OR_FAIL(auto from_combined, iceberg::BoundingBox::Deserialize(combined));
+  ASSERT_EQ(from_combined, expected);
+
+  ASSERT_THAT(
+      iceberg::BoundingBox::Deserialize(std::vector<uint8_t>{0x10, 0x00, 0x00, 0x00}),
+      IsError(iceberg::ErrorKind::kInvalidArgument));
+  combined.pop_back();
+  ASSERT_THAT(iceberg::BoundingBox::Deserialize(combined),
+              IsError(iceberg::ErrorKind::kInvalidArgument));
+}
+
+iceberg::BoundingBox MakeXYBoundingBox(double xmin, double ymin, double xmax,
+                                       double ymax) {
+  return {iceberg::GeospatialBound::XY(xmin, ymin),
+          iceberg::GeospatialBound::XY(xmax, ymax)};
+}
+
+void AssertIntersects(const iceberg::Type& type, const iceberg::BoundingBox& first,
+                      const iceberg::BoundingBox& second, bool expected) {
+  ASSERT_EQ(iceberg::GeospatialBoundsIntersect(type, first, second), expected);
+}
+
+TEST(GeospatialBoundsIntersectTest, TwoDimensions) {
+  const auto geometry = iceberg::geometry();
+  const auto first = MakeXYBoundingBox(0.0, 0.0, 10.0, 10.0);
+  AssertIntersects(*geometry, first, MakeXYBoundingBox(10.0, 2.0, 12.0, 4.0), true);
+  AssertIntersects(*geometry, first, MakeXYBoundingBox(11.0, 2.0, 12.0, 4.0), false);
+
+  const auto geography = iceberg::geography();
+  const auto wrapped = MakeXYBoundingBox(170.0, -10.0, -170.0, 10.0);
+  const std::array wrapped_cases = {
+      std::pair{MakeXYBoundingBox(175.0, -5.0, 178.0, 5.0), true},
+      std::pair{MakeXYBoundingBox(-10.0, -5.0, 10.0, 5.0), false},
+      std::pair{MakeXYBoundingBox(160.0, -5.0, -175.0, 5.0), true},
+  };
+  for (const auto& [candidate, expected] : wrapped_cases) {
+    AssertIntersects(*geography, wrapped, candidate, expected);
+  }
+
+  const auto full_wrap = MakeXYBoundingBox(180.0, -5.0, -180.0, 5.0);
+  AssertIntersects(*geography, full_wrap, MakeXYBoundingBox(-180.0, -5.0, -180.0, 5.0),
+                   true);
+  AssertIntersects(*geography, full_wrap, MakeXYBoundingBox(180.0, -5.0, 180.0, 5.0),
+                   true);
+}
+
+TEST(GeospatialBoundsIntersectTest, OptionalDimensions) {
+  iceberg::BoundingBox xyz1(iceberg::GeospatialBound::XYZ(0.0, 0.0, 0.0),
+                            iceberg::GeospatialBound::XYZ(10.0, 10.0, 1.0));
+  iceberg::BoundingBox xyz2(iceberg::GeospatialBound::XYZ(0.0, 0.0, 2.0),
+                            iceberg::GeospatialBound::XYZ(10.0, 10.0, 3.0));
+  iceberg::BoundingBox xy(iceberg::GeospatialBound::XY(0.0, 0.0),
+                          iceberg::GeospatialBound::XY(10.0, 10.0));
+  const auto geometry = iceberg::geometry();
+  AssertIntersects(*geometry, xyz1, xyz2, false);
+  AssertIntersects(*geometry, xyz1, xy, true);
+
+  iceberg::BoundingBox xym1(iceberg::GeospatialBound::XYM(0.0, 0.0, 0.0),
+                            iceberg::GeospatialBound::XYM(10.0, 10.0, 1.0));
+  iceberg::BoundingBox xym2(iceberg::GeospatialBound::XYM(0.0, 0.0, 2.0),
+                            iceberg::GeospatialBound::XYM(10.0, 10.0, 3.0));
+  AssertIntersects(*geometry, xym1, xym2, false);
+
+  iceberg::BoundingBox xyzm1(iceberg::GeospatialBound::XYZM(0.0, 0.0, 0.0, 0.0),
+                             iceberg::GeospatialBound::XYZM(10.0, 10.0, 2.0, 2.0));
+  iceberg::BoundingBox xyzm2(iceberg::GeospatialBound::XYZM(1.0, 1.0, 1.0, 1.0),
+                             iceberg::GeospatialBound::XYZM(11.0, 11.0, 3.0, 3.0));
+  iceberg::BoundingBox xyzm_disjoint_m(
+      iceberg::GeospatialBound::XYZM(1.0, 1.0, 1.0, 3.0),
+      iceberg::GeospatialBound::XYZM(11.0, 11.0, 3.0, 4.0));
+  AssertIntersects(*geometry, xyzm1, xyzm2, true);
+  AssertIntersects(*geometry, xyzm1, xyzm_disjoint_m, false);
+}
+
+TEST(GeospatialBoundsIntersectTest, RejectsInvalidArguments) {
+  iceberg::BoundingBox valid(iceberg::GeospatialBound::XYZM(0.0, 0.0, 0.0, 0.0),
+                             iceberg::GeospatialBound::XYZM(10.0, 10.0, 1.0, 1.0));
+  auto invalid_x = MakeXYBoundingBox(2.0, 0.0, 1.0, 1.0);
+  ASSERT_THAT(iceberg::GeospatialBoundsIntersect(*iceberg::geometry(), invalid_x, valid),
+              IsError(iceberg::ErrorKind::kInvalidArgument));
+
+  const std::array invalid_geography_bounds = {
+      MakeXYBoundingBox(-181.0, -10.0, 0.0, 10.0),
+      MakeXYBoundingBox(0.0, -91.0, 10.0, 10.0),
+  };
+  for (const auto& invalid : invalid_geography_bounds) {
+    ASSERT_THAT(iceberg::GeospatialBoundsIntersect(*iceberg::geography(), invalid, valid),
+                IsError(iceberg::ErrorKind::kInvalidArgument));
+  }
+
+  iceberg::BoundingBox invalid_z(iceberg::GeospatialBound::XYZ(0.0, 0.0, 2.0),
+                                 iceberg::GeospatialBound::XYZ(10.0, 10.0, 1.0));
+  iceberg::BoundingBox invalid_m(iceberg::GeospatialBound::XYM(0.0, 0.0, 2.0),
+                                 iceberg::GeospatialBound::XYM(10.0, 10.0, 1.0));
+  const std::array<std::shared_ptr<iceberg::Type>, 2> geospatial_types = {
+      iceberg::geometry(), iceberg::geography()};
+  for (const auto& type : geospatial_types) {
+    for (const auto& invalid : {invalid_z, invalid_m}) {
+      ASSERT_THAT(iceberg::GeospatialBoundsIntersect(*type, invalid, valid),
+                  IsError(iceberg::ErrorKind::kInvalidArgument));
+    }
+  }
+
+  ASSERT_THAT(iceberg::GeospatialBoundsIntersect(*iceberg::string(), valid, valid),
+              IsError(iceberg::ErrorKind::kNotSupported));
 }
 
 TEST(TypeTest, Decimal) {
diff --git a/src/iceberg/test/visit_type_test.cc b/src/iceberg/test/visit_type_test.cc
index a6bd9f8c6..8e3187c40 100644
--- a/src/iceberg/test/visit_type_test.cc
+++ b/src/iceberg/test/visit_type_test.cc
@@ -193,14 +193,14 @@ const static std::array<TypeTestCase, 21> kPrimitiveTypes = {{
         .type = iceberg::geometry(),
         .type_id = iceberg::TypeId::kGeometry,
         .primitive = true,
-        .repr = "geometry",
+        .repr = "geometry(OGC:CRS84)",
     },
     {
         .name = "geography",
         .type = iceberg::geography(),
         .type_id = iceberg::TypeId::kGeography,
         .primitive = true,
-        .repr = "geography",
+        .repr = "geography(OGC:CRS84, spherical)",
     },
 }};
 
diff --git a/src/iceberg/type.cc b/src/iceberg/type.cc
index fe48e6a99..488f9615c 100644
--- a/src/iceberg/type.cc
+++ b/src/iceberg/type.cc
@@ -379,28 +379,17 @@ Result<std::unique_ptr<GeometryType>> GeometryType::Make(std::string crs) {
   return std::unique_ptr<GeometryType>(new GeometryType(std::move(crs)));
 }
 
-GeometryType::GeometryType(std::string crs) {
-  if (StringUtils::ToLower(crs) != StringUtils::ToLower(kDefaultCrs)) {
-    crs_ = std::move(crs);
-  }
-}
+GeometryType::GeometryType(std::string crs) : crs_(std::move(crs)) {}
 
-std::string_view GeometryType::crs() const {
-  return crs_.empty() ? kDefaultCrs : std::string_view(crs_);
-}
+std::string_view GeometryType::crs() const { return crs_; }
 TypeId GeometryType::type_id() const { return kTypeId; }
-std::string GeometryType::ToString() const {
-  if (crs_.empty()) {
-    return "geometry";
-  }
-  return std::format("geometry({})", crs_);
-}
+std::string GeometryType::ToString() const { return std::format("geometry({})", crs_); }
 bool GeometryType::Equals(const Type& other) const {
   if (other.type_id() != kTypeId) {
     return false;
   }
   const auto& geometry = static_cast<const GeometryType&>(other);
-  return crs_ == geometry.crs_;
+  return StringUtils::EqualsIgnoreCase(crs_, geometry.crs_);
 }
 
 Result<std::unique_ptr<GeographyType>> GeographyType::Make() {
@@ -422,41 +411,24 @@ Result<std::unique_ptr<GeographyType>> GeographyType::Make(std::string crs,
   return std::unique_ptr<GeographyType>(new GeographyType(std::move(crs), algorithm));
 }
 
-GeographyType::GeographyType(std::string crs) {
-  if (StringUtils::ToLower(crs) != StringUtils::ToLower(kDefaultCrs)) {
-    crs_ = std::move(crs);
-  }
-}
+GeographyType::GeographyType(std::string crs) : crs_(std::move(crs)) {}
 
 GeographyType::GeographyType(std::string crs, EdgeAlgorithm algorithm)
-    : algorithm_(algorithm) {
-  if (StringUtils::ToLower(crs) != StringUtils::ToLower(kDefaultCrs)) {
-    crs_ = std::move(crs);
-  }
-}
+    : crs_(std::move(crs)), algorithm_(algorithm) {}
 
-std::string_view GeographyType::crs() const {
-  return crs_.empty() ? kDefaultCrs : std::string_view(crs_);
-}
-EdgeAlgorithm GeographyType::algorithm() const {
-  return algorithm_.value_or(kDefaultAlgorithm);
-}
+std::string_view GeographyType::crs() const { return crs_; }
+EdgeAlgorithm GeographyType::algorithm() const { return algorithm_; }
 TypeId GeographyType::type_id() const { return kTypeId; }
 std::string GeographyType::ToString() const {
-  if (algorithm_.has_value()) {
-    return std::format("geography({}, {})", crs(), iceberg::ToString(*algorithm_));
-  }
-  if (!crs_.empty()) {
-    return std::format("geography({})", crs_);
-  }
-  return "geography";
+  return std::format("geography({}, {})", crs_, iceberg::ToString(algorithm_));
 }
 bool GeographyType::Equals(const Type& other) const {
   if (other.type_id() != kTypeId) {
     return false;
   }
   const auto& geography = static_cast<const GeographyType&>(other);
-  return crs_ == geography.crs_ && algorithm_ == geography.algorithm_;
+  return StringUtils::EqualsIgnoreCase(crs_, geography.crs_) &&
+         algorithm_ == geography.algorithm_;
 }
 
 FixedType::FixedType(int32_t length) : length_(length) {
diff --git a/src/iceberg/type.h b/src/iceberg/type.h
index 41484333d..dc225c266 100644
--- a/src/iceberg/type.h
+++ b/src/iceberg/type.h
@@ -583,7 +583,7 @@ class ICEBERG_EXPORT GeometryType : public PrimitiveType {
   GeometryType() = default;
   explicit GeometryType(std::string crs);
 
-  std::string crs_;
+  std::string crs_{kDefaultCrs};
 };
 
 /// \brief A data type representing OGC geography in WKB format.
@@ -613,8 +613,8 @@ class ICEBERG_EXPORT GeographyType : public PrimitiveType {
   explicit GeographyType(std::string crs);
   GeographyType(std::string crs, EdgeAlgorithm algorithm);
 
-  std::string crs_;
-  std::optional<EdgeAlgorithm> algorithm_;
+  std::string crs_{kDefaultCrs};
+  EdgeAlgorithm algorithm_{kDefaultAlgorithm};
 };
 
 /// @}