diff --git a/tea/common/iceberg_json.cpp b/tea/common/iceberg_json.cpp
index e733db4..e4dc2b0 100644
--- a/tea/common/iceberg_json.cpp
+++ b/tea/common/iceberg_json.cpp
@@ -31,6 +31,7 @@ constexpr std::string_view kNameField = "name";
 constexpr std::string_view kTypeField = "type";
 constexpr std::string_view kFieldsField = "type";
 constexpr std::string_view kSchemaField = "schema";
+constexpr std::string_view kSchemaNameMappingField = "schema_name_mapping";
 constexpr std::string_view kScanMetadata = "scan_metadata";
 constexpr std::string_view kScanMetadataIdentifier = "scan_metadata_identifier";
 constexpr std::string_view kPartitionsField = "partitions";
@@ -204,6 +205,9 @@ rapidjson::Value Serializer::Serialize(iceberg::ice_tea::ScanMetadata&& scan_met
   rapidjson::Value result(rapidjson::kObjectType);
   AddMember(result, kSchemaField, std::move(*scan_metadata.schema));
   AddMember(result, kPartitionsField, std::move(scan_metadata.partitions));
+  if (scan_metadata.schema_name_mapping.has_value()) {
+    AddMember(result, kSchemaNameMappingField, std::move(*scan_metadata.schema_name_mapping));
+  }
   return result;
 }
 
@@ -375,7 +379,13 @@ iceberg::ice_tea::ScanMetadata Deserialize(const rapidjson::Value& value) {
 
   auto schema = std::make_shared<iceberg::Schema>(Extract<iceberg::Schema>(value, kSchemaField));
   auto partitions = Extract<std::vector<iceberg::ice_tea::ScanMetadata::Partition>>(value, kPartitionsField);
-  return iceberg::ice_tea::ScanMetadata{.schema = std::move(schema), .partitions = std::move(partitions)};
+  std::optional<std::string> schema_name_mapping;
+  if (value.HasMember(kSchemaNameMappingField.data())) {
+    schema_name_mapping = Extract<std::string>(value, kSchemaNameMappingField);
+  }
+  return iceberg::ice_tea::ScanMetadata{.schema = std::move(schema),
+                                        .partitions = std::move(partitions),
+                                        .schema_name_mapping = std::move(schema_name_mapping)};
 }
 
 template <>
diff --git a/tea/common/utils.cpp b/tea/common/utils.cpp
index 28422be..72047fc 100644
--- a/tea/common/utils.cpp
+++ b/tea/common/utils.cpp
@@ -914,6 +914,7 @@ iceberg::ice_tea::ScanMetadata SplitPartitionsAndFilter(iceberg::ice_tea::ScanMe
   const auto& partitions = scan_metadata.partitions;
   iceberg::ice_tea::ScanMetadata result;
   result.schema = scan_metadata.schema;
+  result.schema_name_mapping = scan_metadata.schema_name_mapping;
 
   int iterator = 0;
   for (const auto& partition : partitions) {
diff --git a/tea/gpext/tea_reader.cpp b/tea/gpext/tea_reader.cpp
index acfbc32..790fd73 100644
--- a/tea/gpext/tea_reader.cpp
+++ b/tea/gpext/tea_reader.cpp
@@ -971,6 +971,7 @@ std::shared_ptr<tea::samovar::SingleQueueClient> SamovarMakePlan(TeaContextPtr t
 
     {
       std::shared_ptr<iceberg::Schema> schema = tea::GetSchemaForSnapshot(table_metadata, config.snapshot_ref);
+      std::optional<std::string> schema_name_mapping = tea::meta::access::GetSchemaNameMappingDefault(*table_metadata);
       TEA_LOG("Samovar: getting manifest files");
 
       std::deque<iceberg::ManifestFile> manifest_files_queue =
@@ -987,8 +988,8 @@ std::shared_ptr<tea::samovar::SingleQueueClient> SamovarMakePlan(TeaContextPtr t
             CreateSamovarClient(tea_ctx, queue_name, query_scans_count_key, segment_id, segment_count,
                                 tea::samovar::SamovarRole::kFollower);
         TEA_LOG("Samovar: filling manifests queue");
-        auto maybe_stats = tea::samovar::FillSamovarWithManifests(get::Config(tea_ctx), schema, manifest_files_queue,
-                                                                  segment_count, samovar_client);
+        auto maybe_stats = tea::samovar::FillSamovarWithManifests(get::Config(tea_ctx), schema, schema_name_mapping,
+                                                                  manifest_files_queue, segment_count, samovar_client);
         get::PlannerStats(tea_ctx).Combine(iceberg::ValueSafe(maybe_stats));
 
         return samovar_client;
@@ -1024,6 +1025,7 @@ std::shared_ptr<tea::samovar::SingleQueueClient> SamovarMakePlan(TeaContextPtr t
 
         iceberg::ice_tea::ScanMetadata all_meta =
             iceberg::ValueSafe(iceberg::ice_tea::GetScanMetadata(*entries_stream, *table_metadata, schema, logger));
+        all_meta.schema_name_mapping = schema_name_mapping;
 
         ValidateAllMetadata(get::Config(tea_ctx), all_meta);
 
diff --git a/tea/metadata/access_iceberg.cpp b/tea/metadata/access_iceberg.cpp
index 792efed..872ff97 100644
--- a/tea/metadata/access_iceberg.cpp
+++ b/tea/metadata/access_iceberg.cpp
@@ -102,6 +102,18 @@ std::string GetIcebergTableLocation(const Config& config, TableId table_id) {
   return table->Location();
 }
 
+std::optional<std::string> GetMetadataProperty(const iceberg::TableMetadataV2& table_metadata,
+                                               const std::string& property_name) {
+  if (auto it = table_metadata.properties.find(property_name); it != table_metadata.properties.end()) {
+    return it->second;
+  }
+  return std::nullopt;
+}
+
+std::optional<std::string> GetSchemaNameMappingDefault(const iceberg::TableMetadataV2& table_metadata) {
+  return GetMetadataProperty(table_metadata, "schema.name-mapping.default");
+}
+
 namespace {
 class EmptyIcebergStream : public iceberg::ice_tea::IcebergEntriesStream {
  public:
@@ -177,7 +189,10 @@ std::pair<iceberg::ice_tea::ScanMetadata, PlannerStats> FromIcebergWithLocation(
       }
     }
 
-    return iceberg::ice_tea::GetScanMetadata(*entries_stream, *table_metadata, scan_schema, logger);
+    ARROW_ASSIGN_OR_RAISE(auto scan_metadata,
+                          iceberg::ice_tea::GetScanMetadata(*entries_stream, *table_metadata, scan_schema, logger));
+    scan_metadata.schema_name_mapping = GetSchemaNameMappingDefault(*table_metadata);
+    return scan_metadata;
   }();
 
   if (result.ok()) {
diff --git a/tea/metadata/access_iceberg.h b/tea/metadata/access_iceberg.h
index 272daa7..66a8621 100644
--- a/tea/metadata/access_iceberg.h
+++ b/tea/metadata/access_iceberg.h
@@ -1,7 +1,9 @@
 #pragma once
 
 #include <memory>
+#include <optional>
 #include <string>
+#include <string_view>
 #include <utility>
 
 #include "iceberg/common/fs/filesystem_provider.h"
@@ -29,4 +31,9 @@ std::pair<iceberg::ice_tea::ScanMetadata, PlannerStats> FromIcebergWithLocation(
 
 std::string GetIcebergTableLocation(const Config& config, TableId table_id);
 
+std::optional<std::string> GetMetadataProperty(const iceberg::TableMetadataV2& table_metadata,
+                                               const std::string& property_name);
+
+std::optional<std::string> GetSchemaNameMappingDefault(const iceberg::TableMetadataV2& table_metadata);
+
 }  // namespace tea::meta::access
diff --git a/tea/metadata/planner.h b/tea/metadata/planner.h
index 7557c1e..c086c9c 100644
--- a/tea/metadata/planner.h
+++ b/tea/metadata/planner.h
@@ -13,6 +13,7 @@ namespace tea::meta {
 
 class PlannedMeta {
  public:
+  // this field also contains schema and schema_name_mapping. TODO(gmusya): refactor and rename
   iceberg::ice_tea::ScanMetadata& GetDeletes() { return metadata_; }
   std::shared_ptr<AnnotatedDataEntryStream> GetStream() { return data_meta_stream_; }
 
diff --git a/tea/reader.cpp b/tea/reader.cpp
index 6a2fd64..236dade 100644
--- a/tea/reader.cpp
+++ b/tea/reader.cpp
@@ -578,7 +578,9 @@ class LowercaseRenamingStream : public iceberg::IcebergStream {
 };
 
 arrow::Status Reader::Plan(meta::PlannedMeta meta, const Reader::SerializedFilter& filter, bool postfilter_on_gp) {
-  schema_ = meta.GetDeletes().schema;
+  const auto& deletes_with_schema = meta.GetDeletes();
+  schema_ = deletes_with_schema.schema;
+  auto schema_name_mapping = deletes_with_schema.schema_name_mapping;
   iceberg::Ensure(schema_ != nullptr, std::string(__PRETTY_FUNCTION__) + ": schema is nullptr");
 
   // empty table
@@ -603,8 +605,8 @@ arrow::Status Reader::Plan(meta::PlannedMeta meta, const Reader::SerializedFilte
   auto equality_deletes = std::make_shared<iceberg::EqualityDeletes>(iceberg::EqualityDeletes{});
   iceberg::EqualityDeleteHandler::Config equality_delete_config = MakeIcebergEqualityDeleteConfig(config_);
 
-  for (size_t partition_id = 0; partition_id < meta.GetDeletes().partitions.size(); ++partition_id) {
-    auto& partition = meta.GetDeletes().partitions.at(partition_id);
+  for (size_t partition_id = 0; partition_id < deletes_with_schema.partitions.size(); ++partition_id) {
+    auto& partition = deletes_with_schema.partitions.at(partition_id);
     for (size_t layer_id = 0; layer_id < partition.size(); ++layer_id) {
       auto& layer = partition[layer_id];
       if (!layer.positional_delete_entries_.empty()) {
@@ -699,7 +701,8 @@ arrow::Status Reader::Plan(meta::PlannedMeta meta, const Reader::SerializedFilte
 
   stream_ = iceberg::IcebergScanBuilder::MakeIcebergStream(
       meta_stream, std::move(positional_deletes), std::move(equality_deletes), std::move(equality_delete_config),
-      rg_filter, ice_filter, *schema_, std::move(field_ids_to_retrieve), file_reader_provider, std::nullopt, logger_);
+      rg_filter, ice_filter, *schema_, std::move(field_ids_to_retrieve), file_reader_provider, schema_name_mapping,
+      logger_);
 
   stream_ = std::make_shared<LowercaseRenamingStream>(stream_);
   return arrow::Status::OK();
diff --git a/tea/samovar/planner.cpp b/tea/samovar/planner.cpp
index ce5347b..7fe8f0b 100644
--- a/tea/samovar/planner.cpp
+++ b/tea/samovar/planner.cpp
@@ -76,13 +76,14 @@ std::shared_ptr<SingleQueueClient> MakeSamovarDataClient(const SamovarConfig& co
 }
 
 arrow::Result<PlannerStats> FillSamovarWithManifests(const Config& config, std::shared_ptr<iceberg::Schema> schema,
+                                                     std::optional<std::string> schema_name_mapping,
                                                      std::deque<iceberg::ManifestFile> manifests, int segment_count,
                                                      std::shared_ptr<SingleQueueClient> samovar_client) {
   PlannerStats stats;
   std::optional<ScopedTimerTicks> timer = ScopedTimerTicks(stats.plan_duration);
 
   samovar::ScanMetadata result;
-  *result.mutable_schema() = IcebergSchemaToTeapotSchema(schema);
+  AttachSchema(result, schema, schema_name_mapping);
 
   std::vector<samovar::ManifestList> samovar_manifests = ConvertToSamovarManifestLists(manifests);
 
@@ -361,7 +362,7 @@ arrow::Result<std::pair<meta::PlannedMeta, PlannerStats>> FromSamovar(
   auto response = samovar_client->GetPlannedMetadata();
   if (response.scan_already_finished()) {
     iceberg::ice_tea::ScanMetadata metadata;
-    metadata.schema = TeapotSchemaToIcebergSchema(response.schema());
+    AttachSchema(metadata, response);
 
     auto sched = std::make_shared<EmptyMetadataScheduler>();
 
@@ -397,7 +398,7 @@ arrow::Result<std::pair<meta::PlannedMeta, PlannerStats>> FromSamovar(
     samovar_client->WaitForManifestsQueue();
 
     iceberg::ice_tea::ScanMetadata metadata;
-    metadata.schema = TeapotSchemaToIcebergSchema(response.schema());
+    AttachSchema(metadata, response);
 
     // process tasks from the entries queue
     auto sched = std::make_shared<SamovarMetadataScheduler>(config, samovar_client);
diff --git a/tea/samovar/planner.h b/tea/samovar/planner.h
index 3aee48f..ca44076 100644
--- a/tea/samovar/planner.h
+++ b/tea/samovar/planner.h
@@ -5,6 +5,7 @@
 
 #include <deque>
 #include <memory>
+#include <optional>
 #include <string>
 #include <utility>
 
@@ -28,6 +29,7 @@ arrow::Result<PlannerStats> FillSamovar(const Config& config, iceberg::ice_tea::
                                         std::shared_ptr<SingleQueueClient> samovar_client);
 
 arrow::Result<PlannerStats> FillSamovarWithManifests(const Config& config, std::shared_ptr<iceberg::Schema> schema,
+                                                     std::optional<std::string> schema_name_mapping,
                                                      std::deque<iceberg::ManifestFile>, int segment_count,
                                                      std::shared_ptr<SingleQueueClient> samovar_client);
 
diff --git a/tea/samovar/proto/samovar.proto b/tea/samovar/proto/samovar.proto
index 8388c7f..4ad29e9 100644
--- a/tea/samovar/proto/samovar.proto
+++ b/tea/samovar/proto/samovar.proto
@@ -55,6 +55,8 @@ message ScanMetadata {
 
   bool use_distributed_metadata_processing = 3;
   bool scan_already_finished = 4;
+
+  optional string schema_name_mapping = 5;
 }
 
 message AnnotatedDataEntry {
diff --git a/tea/samovar/utils.cpp b/tea/samovar/utils.cpp
index a1217e3..5417e77 100644
--- a/tea/samovar/utils.cpp
+++ b/tea/samovar/utils.cpp
@@ -71,7 +71,8 @@ samovar::ScanMetadata ConvertIcebergScanMetaToSamovarRepresentation(iceberg::ice
   };
 
   samovar::ScanMetadata result;
-  *result.mutable_schema() = IcebergSchemaToTeapotSchema(scan_metadata.schema);
+  AttachSchema(result, scan_metadata.schema, scan_metadata.schema_name_mapping);
+
   for (const auto& partition : scan_metadata.partitions) {
     auto* samovar_partition = result.add_partitions();
     for (const auto& layer : partition) {
@@ -139,7 +140,7 @@ iceberg::ice_tea::ScanMetadata ConvertSamovarRepresentationToScanMeta(const samo
     result.partitions.push_back(std::move(ice_partition));
   }
 
-  result.schema = TeapotSchemaToIcebergSchema(scan_metadata.schema());
+  AttachSchema(result, scan_metadata);
 
   return result;
 }
@@ -475,6 +476,9 @@ void SendManifestLists(const std::shared_ptr<ISamovarClient> client,
 samovar::ScanMetadata ClearDataEntries(const samovar::ScanMetadata& scan_metadata) {
   samovar::ScanMetadata result;
   *result.mutable_schema() = scan_metadata.schema();
+  if (scan_metadata.has_schema_name_mapping()) {
+    result.set_schema_name_mapping(scan_metadata.schema_name_mapping());
+  }
   result.set_use_distributed_metadata_processing(scan_metadata.use_distributed_metadata_processing());
   for (int i = 0; i < scan_metadata.partitions_size(); ++i) {
     const auto& partition = scan_metadata.partitions()[i];
diff --git a/tea/samovar/utils.h b/tea/samovar/utils.h
index 9f92335..dddf073 100644
--- a/tea/samovar/utils.h
+++ b/tea/samovar/utils.h
@@ -11,6 +11,7 @@
 #include "iceberg/tea_scan.h"
 
 #include "tea/common/config.h"
+#include "tea/common/utils.h"
 #include "tea/metadata/metadata.h"
 #include "tea/samovar/network_layer/samovar_client.h"
 #include "tea/samovar/proto/samovar.pb.h"
@@ -54,6 +55,21 @@ void SendDataEntries(const std::shared_ptr<ISamovarClient> client,
                      const std::vector<samovar::AnnotatedDataEntry>& additional_data_entries,
                      const std::string& queue_id, std::chrono::seconds ttl_seconds, uint32_t batch_size);
 
+inline void AttachSchema(iceberg::ice_tea::ScanMetadata& result, const samovar::ScanMetadata& source) {
+  result.schema = TeapotSchemaToIcebergSchema(source.schema());
+  if (source.has_schema_name_mapping()) {
+    result.schema_name_mapping = source.schema_name_mapping();
+  }
+}
+
+inline void AttachSchema(samovar::ScanMetadata& result, std::shared_ptr<iceberg::Schema> schema,
+                         const std::optional<std::string>& schema_name_mapping) {
+  *result.mutable_schema() = IcebergSchemaToTeapotSchema(schema);
+  if (schema_name_mapping.has_value()) {
+    result.set_schema_name_mapping(*schema_name_mapping);
+  }
+}
+
 samovar::ScanMetadata ClearDataEntries(const samovar::ScanMetadata& scan_metadata);
 
 }  // namespace tea::samovar
diff --git a/tea/smoke_test/CMakeLists.txt b/tea/smoke_test/CMakeLists.txt
index f3dd422..f916a4d 100644
--- a/tea/smoke_test/CMakeLists.txt
+++ b/tea/smoke_test/CMakeLists.txt
@@ -63,6 +63,7 @@ set(smoke_test_sources
   real_parquet_test.cpp
   row_group_filter_test.cpp
   mapping_test.cpp
+  name_mapping_default_test.cpp
   self_join_test.cpp
   large_metadata_test.cpp
   mismatching_types_test.cpp
diff --git a/tea/smoke_test/name_mapping_default_test.cpp b/tea/smoke_test/name_mapping_default_test.cpp
new file mode 100644
index 0000000..7cd98f0
--- /dev/null
+++ b/tea/smoke_test/name_mapping_default_test.cpp
@@ -0,0 +1,129 @@
+#include <map>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "gtest/gtest.h"
+#include "iceberg/schema.h"
+#include "iceberg/test_utils/assertions.h"
+#include "iceberg/type.h"
+
+#include "tea/smoke_test/environment.h"
+#include "tea/smoke_test/pq.h"
+#include "tea/smoke_test/test_base.h"
+#include "tea/test_utils/column.h"
+#include "tea/test_utils/metadata.h"
+
+namespace tea {
+namespace {
+
+constexpr int kNoFieldId = -1;
+
+class NameMappingDefaultTest : public TeaTest {
+ public:
+  void SetUp() override {
+    if (Environment::GetMetadataType() != MetadataType::kIceberg) {
+      GTEST_SKIP() << "schema.name-mapping.default only applies to the Iceberg metadata path";
+    }
+    TeaTest::SetUp();
+  }
+
+ protected:
+  static std::shared_ptr<iceberg::Schema> MakeIcebergSchema() {
+    using iceberg::types::NestedField;
+    using iceberg::types::PrimitiveType;
+
+    std::vector<NestedField> fields;
+    fields.push_back(NestedField{.name = "col1",
+                                 .field_id = 1,
+                                 .is_required = false,
+                                 .type = std::make_shared<PrimitiveType>(iceberg::TypeID::kString)});
+    fields.push_back(NestedField{.name = "col2",
+                                 .field_id = 2,
+                                 .is_required = false,
+                                 .type = std::make_shared<PrimitiveType>(iceberg::TypeID::kInt)});
+    fields.push_back(NestedField{.name = "col3",
+                                 .field_id = 3,
+                                 .is_required = false,
+                                 .type = std::make_shared<PrimitiveType>(iceberg::TypeID::kLong)});
+    return std::make_shared<iceberg::Schema>(0, std::move(fields));
+  }
+};
+
+TEST_F(NameMappingDefaultTest, ResolvesFieldIdsByName) {
+  std::string s0 = "a";
+  std::string s1 = "b";
+
+  auto col1 = MakeStringColumn("col1", kNoFieldId, std::vector<std::string*>{&s0, &s1});
+  auto col2 = MakeInt32Column("col2", kNoFieldId, OptionalVector<int32_t>{10, 20});
+  auto col3 = MakeInt64Column("col3", kNoFieldId, OptionalVector<int64_t>{100, 200});
+
+  ASSIGN_OR_FAIL(auto data_path, state_->WriteFile({col1, col2, col3}));
+  ASSERT_OK(state_->AddDataFiles({data_path}));
+
+  state_->SetSchema(MakeIcebergSchema());
+  state_->SetProperties({{"schema.name-mapping.default", R"([{"field-id":1,"names":["col1"]},)"
+                                                         R"({"field-id":2,"names":["col2"]},)"
+                                                         R"({"field-id":3,"names":["col3"]}])"}});
+
+  ASSIGN_OR_FAIL(auto defer, state_->CreateTable({GreenplumColumnInfo{.name = "col1", .type = "text"},
+                                                  GreenplumColumnInfo{.name = "col2", .type = "int4"},
+                                                  GreenplumColumnInfo{.name = "col3", .type = "int8"}}));
+
+  ASSIGN_OR_FAIL(auto result, pq::TableScanQuery(kDefaultTableName, "col1, col2, col3").Run(*conn_));
+
+  pq::ScanResult expected_result({"col1", "col2", "col3"}, {{"a", "10", "100"}, {"b", "20", "200"}});
+  EXPECT_EQ(result, expected_result);
+}
+
+TEST_F(NameMappingDefaultTest, ParquetNameDiffersFromIcebergName) {
+  std::string s0 = "a";
+  std::string s1 = "b";
+
+  auto col1 = MakeStringColumn("old_col1", kNoFieldId, std::vector<std::string*>{&s0, &s1});
+  auto col2 = MakeInt32Column("old_col2", kNoFieldId, OptionalVector<int32_t>{10, 20});
+  auto col3 = MakeInt64Column("old_col3", kNoFieldId, OptionalVector<int64_t>{100, 200});
+
+  ASSIGN_OR_FAIL(auto data_path, state_->WriteFile({col1, col2, col3}));
+  ASSERT_OK(state_->AddDataFiles({data_path}));
+
+  state_->SetSchema(MakeIcebergSchema());
+  state_->SetProperties({{"schema.name-mapping.default", R"([{"field-id":1,"names":["old_col1"]},)"
+                                                         R"({"field-id":2,"names":["old_col2"]},)"
+                                                         R"({"field-id":3,"names":["old_col3"]}])"}});
+
+  ASSIGN_OR_FAIL(auto defer, state_->CreateTable({GreenplumColumnInfo{.name = "col1", .type = "text"},
+                                                  GreenplumColumnInfo{.name = "col2", .type = "int4"},
+                                                  GreenplumColumnInfo{.name = "col3", .type = "int8"}}));
+
+  ASSIGN_OR_FAIL(auto result, pq::TableScanQuery(kDefaultTableName, "col1, col2, col3").Run(*conn_));
+
+  pq::ScanResult expected_result({"col1", "col2", "col3"}, {{"a", "10", "100"}, {"b", "20", "200"}});
+  EXPECT_EQ(result, expected_result);
+}
+
+TEST_F(NameMappingDefaultTest, NoFieldIdNoSchemaNameMapping) {
+  std::string s0 = "a";
+  std::string s1 = "b";
+
+  auto col1 = MakeStringColumn("col1", kNoFieldId, std::vector<std::string*>{&s0, &s1});
+  auto col2 = MakeInt32Column("col2", kNoFieldId, OptionalVector<int32_t>{10, 20});
+  auto col3 = MakeInt64Column("col3", kNoFieldId, OptionalVector<int64_t>{100, 200});
+
+  ASSIGN_OR_FAIL(auto data_path, state_->WriteFile({col1, col2, col3}));
+  ASSERT_OK(state_->AddDataFiles({data_path}));
+
+  state_->SetSchema(MakeIcebergSchema());
+
+  ASSIGN_OR_FAIL(auto defer, state_->CreateTable({GreenplumColumnInfo{.name = "col1", .type = "text"},
+                                                  GreenplumColumnInfo{.name = "col2", .type = "int4"},
+                                                  GreenplumColumnInfo{.name = "col3", .type = "int8"}}));
+
+  ASSIGN_OR_FAIL(auto result, pq::TableScanQuery(kDefaultTableName, "col1, col2, col3").Run(*conn_));
+
+  pq::ScanResult expected_result({"col1", "col2", "col3"}, {{"", "", ""}, {"", "", ""}});
+  EXPECT_EQ(result, expected_result);
+}
+
+}  // namespace
+}  // namespace tea
diff --git a/tea/smoke_test/test_base.h b/tea/smoke_test/test_base.h
index 13df0de..9ab4e06 100644
--- a/tea/smoke_test/test_base.h
+++ b/tea/smoke_test/test_base.h
@@ -2,6 +2,7 @@
 
 #include <map>
 #include <memory>
+#include <stdexcept>
 #include <string>
 #include <utility>
 #include <vector>
@@ -78,6 +79,14 @@ class TeapotMetadataWriter : public IMetadataWriter {
                                    Options{.profile = Environment::GetProfile()}));
   }
 
+  void SetSchema(std::shared_ptr<iceberg::Schema>) override {
+    throw std::runtime_error("Internal error in test. SetSchema is not supported for TeapotMetadataWriter");
+  }
+
+  void SetProperties(std::map<std::string, std::string>) override {
+    throw std::runtime_error("Internal error in test. SetSchema is not supported for TeapotMetadataWriter");
+  }
+
  private:
   const TableName table_name_;
   std::vector<FragmentInfo> fragments_;
@@ -163,6 +172,16 @@ class TestState {
     return table_creator_->CreateTable(column_infos, table_name, location);
   }
 
+  void SetSchema(std::shared_ptr<iceberg::Schema> schema, const TableName& table_name = kDefaultTableName) {
+    BuildMetadataWriterIfNecessary(table_name);
+    metadata_writer_.at(table_name)->SetSchema(std::move(schema));
+  }
+
+  void SetProperties(std::map<std::string, std::string> properties, const TableName& table_name = kDefaultTableName) {
+    BuildMetadataWriterIfNecessary(table_name);
+    metadata_writer_.at(table_name)->SetProperties(std::move(properties));
+  }
+
   void SetFileWriter(std::shared_ptr<IFileWriter> file_writer) { file_writer_ = file_writer; }
   void SetMetadataWriterBuilder(std::shared_ptr<IMetadataWriterBuilder> metadata_writer_builder) {
     metadata_writer_.clear();
diff --git a/tea/test_utils/metadata.h b/tea/test_utils/metadata.h
index 1e0f986..c2500eb 100644
--- a/tea/test_utils/metadata.h
+++ b/tea/test_utils/metadata.h
@@ -2,6 +2,7 @@
 
 #include <cstdint>
 #include <cstdlib>
+#include <map>
 #include <memory>
 #include <string>
 #include <utility>
@@ -43,6 +44,9 @@ class IMetadataWriter {
                                                const std::vector<int32_t>& field_ids) = 0;
   virtual arrow::Result<Location> Finalize() = 0;
 
+  virtual void SetSchema(std::shared_ptr<iceberg::Schema>) = 0;
+  virtual void SetProperties(std::map<std::string, std::string>) = 0;
+
   virtual ~IMetadataWriter() = default;
 };
 
@@ -297,7 +301,17 @@ class IcebergMetadataWriter : public IMetadataWriter {
                                          " for column " + column->name());
   }
 
+  void SetSchema(std::shared_ptr<iceberg::Schema> schema) override { forced_schema_ = std::move(schema); }
+
+  void SetProperties(std::map<std::string, std::string> properties) override {
+    table_properties_ = std::move(properties);
+  }
+
   arrow::Result<std::shared_ptr<iceberg::Schema>> GetSchema() const {
+    if (forced_schema_) {
+      return forced_schema_;
+    }
+
     if (!some_data_path_.has_value()) {
       return arrow::Status::ExecutionError("No data file to extract schema");
     }
@@ -379,6 +393,7 @@ class IcebergMetadataWriter : public IMetadataWriter {
         std::make_shared<iceberg::SortOrder>(iceberg::SortOrder{.order_id = 0, .fields = {}}));
     builder.default_sort_order_id = 0;
     builder.partition_specs = {std::make_shared<iceberg::PartitionSpec>(partition_spec_)};
+    builder.properties = table_properties_;
 
     std::shared_ptr<iceberg::TableMetadataV2> meta = builder.Build();
     {
@@ -415,6 +430,9 @@ class IcebergMetadataWriter : public IMetadataWriter {
   const std::string profile_;
   const iceberg::PartitionSpec partition_spec_;
 
+  std::shared_ptr<iceberg::Schema> forced_schema_;
+  std::map<std::string, std::string> table_properties_;
+
   uint64_t data_files_count_ = 0;
   uint64_t delete_files_count_ = 0;
 };
diff --git a/vendor/CMakeLists.txt b/vendor/CMakeLists.txt
index ef2c383..e47283e 100644
--- a/vendor/CMakeLists.txt
+++ b/vendor/CMakeLists.txt
@@ -31,7 +31,10 @@ FetchContent_Declare(
   EXCLUDE_FROM_ALL
   GIT_REPOSITORY ${GITHUB}/lithium-tech/iceberg-cxx.git
   GIT_TAG 3e148ac96d07e9bd3a07dbe9a6ae2efdc6559f2a
+  PATCH_COMMAND git reset --hard HEAD
+    && git apply ${CMAKE_CURRENT_SOURCE_DIR}/iceberg-cxx/0001-scan-metadata-schema-name-mapping.patch
 )
+# TODO(gmusya): support schema.name-mapping.default in GetScanMetadata and add this patch to iceberg-cxx
 
 FetchContent_MakeAvailable(googletest hiredis)
 
diff --git a/vendor/iceberg-cxx/0001-scan-metadata-schema-name-mapping.patch b/vendor/iceberg-cxx/0001-scan-metadata-schema-name-mapping.patch
new file mode 100644
index 0000000..d930522
--- /dev/null
+++ b/vendor/iceberg-cxx/0001-scan-metadata-schema-name-mapping.patch
@@ -0,0 +1,13 @@
+diff --git a/iceberg/tea_scan.h b/iceberg/tea_scan.h
+index 87b0e71..6b658ca 100644
+--- a/iceberg/tea_scan.h
++++ b/iceberg/tea_scan.h
+@@ -98,6 +98,8 @@ struct ScanMetadata {
+   std::shared_ptr<Schema> schema;
+   std::vector<Partition> partitions;
+ 
++  std::optional<std::string> schema_name_mapping;
++
+   bool operator==(const ScanMetadata& scan_meta) const = default;
+ };
+