fix: resolve column-mapped reads per Delta protocol mode

ApplyDeltaColumnMapping previously set DeltaMultiFileColumnDefinition::identifier
unconditionally to the physical name when the kernel reported one, overwriting
any field_id it had just assigned. The kernel emits both `parquet.field.id` and
`delta.columnMapping.physicalName` whenever column mapping is enabled, so the
final identifier was always a string and DuckDB's MultiFileReader always matched
columns by name. That works for spec-conformant writers (pyspark) which name
parquet columns with their physical names, but produces all-null rows in id
mode against parquet files whose columns retain logical names with field_ids
set -- e.g., the typical "rename without rewrite" flow.

Per the Delta protocol "Reader Requirements for Column Mapping":
  - id mode: resolve columns by parquet field_id
  - name mode: resolve columns by physical name
  - none: resolve by display name

Read `delta.columnMapping.mode` from the snapshot's metadata configuration
via the existing kernel FFI `visit_metadata_configuration` and thread the
resolved mode through SchemaVisitor down to ApplyDeltaColumnMapping, which
now sets identifier per spec for the active mode (BIGINT field_id for ID,
VARCHAR physical name for NAME, leaves it unset for NONE so DuckDB matches
by display name). The mode value is lowercased before comparison so a
non-conformant writer's "ID"/"Name" doesn't silently degrade to NONE.

The write-path entry point (VisitWriteContextSchema) is left passing NONE
with a TODO; identifier isn't consumed for column matching during writes,
so this is behavior-preserving.

Adds three inlined SQLLogicTest fixtures, independent of GENERATED_DATA_AVAILABLE:
- column_mapping_id_mode_logical_names: the original bug repro -- id-mode log
  with physical names, parquet file with logical names + field_ids. Fails on
  main, passes with this fix.
- column_mapping_name_mode: name-mode log paired with a parquet file whose
  columns are named with the physical names; locks in that VARCHAR identifier
  dispatch still resolves correctly.
- column_mapping_id_mode_nested_struct: exercises the VisitStruct recursion
  path with field_ids on nested fields, which the top-level fixture does not
  cover. Verified to fail without the fix (returns nested NULLs).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: rymurr

data/inlined/column_mapping_id_mode_logical_names/delta_lake/_delta_log/00000000000000000000.json

@@ -0,0 +1,4 @@
+{"commitInfo":{"timestamp":1700000000000,"operation":"CREATE TABLE","operationParameters":{},"isolationLevel":"Serializable","isBlindAppend":true}}
+{"protocol":{"minReaderVersion":3,"minWriterVersion":7,"readerFeatures":["columnMapping"],"writerFeatures":["columnMapping"]}}
+{"metaData":{"id":"00000000-0000-0000-0000-0000000000ab","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"a\",\"type\":\"long\",\"nullable\":false,\"metadata\":{\"delta.columnMapping.id\":1,\"delta.columnMapping.physicalName\":\"col-a-uuid\"}},{\"name\":\"b\",\"type\":\"string\",\"nullable\":false,\"metadata\":{\"delta.columnMapping.id\":2,\"delta.columnMapping.physicalName\":\"col-b-uuid\"}}]}","partitionColumns":[],"configuration":{"delta.columnMapping.mode":"id","delta.columnMapping.maxColumnId":"2"},"createdTime":1700000000000}}
+{"add":{"path":"part-00000.snappy.parquet","partitionValues":{},"size":869,"modificationTime":1700000000000,"dataChange":true,"stats":"{\"numRecords\":2,\"minValues\":{\"col-a-uuid\":1,\"col-b-uuid\":\"alice\"},\"maxValues\":{\"col-a-uuid\":2,\"col-b-uuid\":\"bob\"},\"nullCount\":{\"col-a-uuid\":0,\"col-b-uuid\":0}}"}}

data/inlined/column_mapping_id_mode_logical_names/delta_lake/part-00000.snappy.parquet

@@ -0,0 +1,4 @@
+{"commitInfo":{"timestamp":1700000000000,"operation":"CREATE TABLE","operationParameters":{},"isolationLevel":"Serializable","isBlindAppend":true}}
+{"protocol":{"minReaderVersion":3,"minWriterVersion":7,"readerFeatures":["columnMapping"],"writerFeatures":["columnMapping"]}}
+{"metaData":{"id":"00000000-0000-0000-0000-0000000000ad","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"nested_col\",\"type\":{\"type\":\"struct\",\"fields\":[{\"name\":\"x\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{\"delta.columnMapping.id\":2,\"delta.columnMapping.physicalName\":\"col-x-uuid\"}},{\"name\":\"y\",\"type\":\"string\",\"nullable\":true,\"metadata\":{\"delta.columnMapping.id\":3,\"delta.columnMapping.physicalName\":\"col-y-uuid\"}}]},\"nullable\":true,\"metadata\":{\"delta.columnMapping.id\":1,\"delta.columnMapping.physicalName\":\"col-nested-uuid\"}}]}","partitionColumns":[],"configuration":{"delta.columnMapping.mode":"id","delta.columnMapping.maxColumnId":"3"},"createdTime":1700000000000}}
+{"add":{"path":"part-00000.snappy.parquet","partitionValues":{},"size":351,"modificationTime":1700000000000,"dataChange":true,"stats":"{\"numRecords\":2}"}}

data/inlined/column_mapping_id_mode_nested_struct/delta_lake/_delta_log/00000000000000000000.json

@@ -0,0 +1,4 @@
+{"commitInfo":{"timestamp":1700000000000,"operation":"CREATE TABLE","operationParameters":{},"isolationLevel":"Serializable","isBlindAppend":true}}
+{"protocol":{"minReaderVersion":3,"minWriterVersion":7,"readerFeatures":["columnMapping"],"writerFeatures":["columnMapping"]}}
+{"metaData":{"id":"00000000-0000-0000-0000-0000000000ac","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"a\",\"type\":\"long\",\"nullable\":false,\"metadata\":{\"delta.columnMapping.id\":1,\"delta.columnMapping.physicalName\":\"col-a-uuid\"}},{\"name\":\"b\",\"type\":\"string\",\"nullable\":false,\"metadata\":{\"delta.columnMapping.id\":2,\"delta.columnMapping.physicalName\":\"col-b-uuid\"}}]}","partitionColumns":[],"configuration":{"delta.columnMapping.mode":"name","delta.columnMapping.maxColumnId":"2"},"createdTime":1700000000000}}
+{"add":{"path":"part-00000.snappy.parquet","partitionValues":{},"size":370,"modificationTime":1700000000000,"dataChange":true,"stats":"{\"numRecords\":2,\"minValues\":{\"col-a-uuid\":1,\"col-b-uuid\":\"alice\"},\"maxValues\":{\"col-a-uuid\":2,\"col-b-uuid\":\"bob\"},\"nullCount\":{\"col-a-uuid\":0,\"col-b-uuid\":0}}"}}

data/inlined/column_mapping_id_mode_nested_struct/delta_lake/part-00000.snappy.parquet

@@ -573,8 +573,9 @@ ffi::EngineSchemaVisitor SchemaVisitor::CreateSchemaVisitor(SchemaVisitor &state
 }
 
 vector<DeltaMultiFileColumnDefinition> SchemaVisitor::VisitSnapshotSchema(ffi::Handle<ffi::SharedExternEngine> engine,
-                                                                          ffi::SharedSnapshot *snapshot) {
-	SchemaVisitor state(engine);
+                                                                          ffi::SharedSnapshot *snapshot,
+                                                                          DeltaColumnMappingMode mapping_mode) {
+	SchemaVisitor state(engine, mapping_mode);
 	auto visitor = CreateSchemaVisitor(state);
 
 	auto schema = logical_schema(snapshot);
@@ -590,8 +591,8 @@ vector<DeltaMultiFileColumnDefinition> SchemaVisitor::VisitSnapshotSchema(ffi::H
 
 vector<DeltaMultiFileColumnDefinition>
 SchemaVisitor::VisitSnapshotGlobalReadSchema(ffi::Handle<ffi::SharedExternEngine> engine, ffi::SharedScan *scan,
-                                             bool logical) {
-	SchemaVisitor visitor_state(engine);
+                                             bool logical, DeltaColumnMappingMode mapping_mode) {
+	SchemaVisitor visitor_state(engine, mapping_mode);
 	auto visitor = CreateSchemaVisitor(visitor_state);
 
 	ffi::Handle<ffi::SharedSchema> schema;
@@ -614,7 +615,10 @@ SchemaVisitor::VisitSnapshotGlobalReadSchema(ffi::Handle<ffi::SharedExternEngine
 vector<DeltaMultiFileColumnDefinition>
 SchemaVisitor::VisitWriteContextSchema(ffi::Handle<ffi::SharedExternEngine> engine,
                                        ffi::SharedWriteContext *write_context) {
-	SchemaVisitor visitor_state(engine);
+	// TODO(column-mapping-writes): plumb the table's column mapping mode here so writes
+	// emit identifiers consistent with the read path. The read path is the only consumer
+	// today, so leaving this NONE keeps writes' behavior unchanged from before this fix.
+	SchemaVisitor visitor_state(engine, DeltaColumnMappingMode::NONE);
 	auto visitor = CreateSchemaVisitor(visitor_state);
 	auto schema = ffi::get_write_schema(write_context);
 	uintptr_t result = visit_schema(schema, &visitor);
@@ -633,7 +637,7 @@ void SchemaVisitor::VisitDecimal(SchemaVisitor *state, uintptr_t sibling_list_id
 	DeltaMultiFileColumnDefinition decimal_def(KernelUtils::FromDeltaString(name), decimal_type, is_nullable);
 	decimal_def.default_expression = make_uniq<ConstantExpression>(Value().DefaultCastAs(decimal_type));
 
-	ApplyDeltaColumnMapping(state->engine, metadata, decimal_def);
+	ApplyDeltaColumnMapping(*state, metadata, decimal_def);
 
 	state->AppendToList(sibling_list_id, name, std::move(decimal_def));
 }
@@ -656,7 +660,7 @@ void SchemaVisitor::VisitStruct(SchemaVisitor *state, uintptr_t sibling_list_id,
 	struct_def.children = std::move(children);
 	struct_def.default_expression = make_uniq<ConstantExpression>(Value(struct_type));
 
-	ApplyDeltaColumnMapping(state->engine, metadata, struct_def);
+	ApplyDeltaColumnMapping(*state, metadata, struct_def);
 
 	state->AppendToList(sibling_list_id, name, std::move(struct_def));
 }
@@ -676,7 +680,7 @@ void SchemaVisitor::VisitArray(SchemaVisitor *state, uintptr_t sibling_list_id,
 	// TODO: kinda wonky, but column mapper uses this
 	list_def.children.front().name = "list";
 
-	ApplyDeltaColumnMapping(state->engine, metadata, list_def);
+	ApplyDeltaColumnMapping(*state, metadata, list_def);
 
 	state->AppendToList(sibling_list_id, name, std::move(list_def));
 }
@@ -699,7 +703,7 @@ void SchemaVisitor::VisitMap(SchemaVisitor *state, uintptr_t sibling_list_id, ff
 
 	map_def.default_expression = make_uniq<ConstantExpression>(Value(map_type));
 
-	ApplyDeltaColumnMapping(state->engine, metadata, map_def);
+	ApplyDeltaColumnMapping(*state, metadata, map_def);
 
 	state->AppendToList(sibling_list_id, name, std::move(map_def));
 }
@@ -711,7 +715,7 @@ void SchemaVisitor::VisitVariant(SchemaVisitor *state, uintptr_t sibling_list_id
 	// global setting.
 	LogicalType type = LogicalType::VARIANT();
 	DeltaMultiFileColumnDefinition col_def(KernelUtils::FromDeltaString(name), type, is_nullable);
-	ApplyDeltaColumnMapping(state->engine, metadata, col_def);
+	ApplyDeltaColumnMapping(*state, metadata, col_def);
 	state->AppendToList(sibling_list_id, name, std::move(col_def));
 }
 
@@ -915,6 +919,30 @@ string KernelUtils::FetchFromStringMap(ffi::Handle<ffi::SharedExternEngine> engi
 	return val;
 }
 
+DeltaColumnMappingMode KernelUtils::ReadColumnMappingMode(ffi::SharedSnapshot *snapshot) {
+	struct VisitorContext {
+		string mode;
+	};
+	VisitorContext ctx;
+	auto visitor = [](ffi::NullableCvoid engine_context, ffi::KernelStringSlice key, ffi::KernelStringSlice value) {
+		auto &c = *static_cast<VisitorContext *>(engine_context);
+		if (FromDeltaString(key) == "delta.columnMapping.mode") {
+			c.mode = FromDeltaString(value);
+		}
+	};
+	ffi::visit_metadata_configuration(snapshot, &ctx, visitor);
+	// The Delta protocol specifies lowercase values, but normalize defensively so a
+	// non-conformant writer's "ID"/"Name" doesn't silently degrade to NONE.
+	auto mode = StringUtil::Lower(ctx.mode);
+	if (mode == "id") {
+		return DeltaColumnMappingMode::ID;
+	}
+	if (mode == "name") {
+		return DeltaColumnMappingMode::NAME;
+	}
+	return DeltaColumnMappingMode::NONE;
+}
+
 vector<unique_ptr<ParsedExpression>> &
 KernelUtils::UnpackTransformExpression(const vector<unique_ptr<ParsedExpression>> &parsed_expression) {
 	if (parsed_expression.size() != 1) {

data/inlined/column_mapping_name_mode/delta_lake/_delta_log/00000000000000000000.json

@@ -620,7 +620,8 @@ void DeltaMultiFileList::Bind(vector<LogicalType> &return_types, vector<string>
 	vector<DeltaMultiFileColumnDefinition> visited_schema;
 	{
 		auto snapshot_ref = snapshot->GetLockingRef();
-		visited_schema = SchemaVisitor::VisitSnapshotSchema(extern_engine.get(), snapshot_ref.GetPtr());
+		auto mapping_mode = KernelUtils::ReadColumnMappingMode(snapshot_ref.GetPtr());
+		visited_schema = SchemaVisitor::VisitSnapshotSchema(extern_engine.get(), snapshot_ref.GetPtr(), mapping_mode);
 	}
 
 	for (const auto &field : visited_schema) {
@@ -791,7 +792,9 @@ void DeltaMultiFileList::InitializeScan() const {
 		}
 	}
 
-	lazy_loaded_schema = SchemaVisitor::VisitSnapshotGlobalReadSchema(extern_engine.get(), scan.get(), true);
+	auto mapping_mode = KernelUtils::ReadColumnMappingMode(snapshot_ref.GetPtr());
+	lazy_loaded_schema =
+	    SchemaVisitor::VisitSnapshotGlobalReadSchema(extern_engine.get(), scan.get(), true, mapping_mode);
 
 	DeltaMultiFileColumnDefinition::Print(lazy_loaded_schema, "lazy_loaded_schema");
 

data/inlined/column_mapping_name_mode/delta_lake/part-00000.snappy.parquet

@@ -48,13 +48,23 @@ struct DeltaLogPathArray {
 	vector<ffi::FfiLogPath> log_entries;
 };
 
+// Mirrors the Delta protocol's `delta.columnMapping.mode` table property.
+// Determines how readers resolve logical columns to parquet columns:
+//   ID   -> resolve by parquet field_id
+//   NAME -> resolve by physical name
+//   NONE -> resolve by display (logical) name
+enum class DeltaColumnMappingMode { NONE, ID, NAME };
+
 struct KernelUtils {
 	static LogicalType GetLogPathType();
 	static ffi::KernelStringSlice ToDeltaString(const string &str);
 	static string FromDeltaString(const struct ffi::KernelStringSlice slice);
 	static vector<bool> FromDeltaBoolSlice(const struct ffi::KernelBoolSlice slice);
 	static string FetchFromStringMap(ffi::Handle<ffi::SharedExternEngine> engine, const ffi::CStringMap *map,
 	                                 const string &key);
+	// Read `delta.columnMapping.mode` from the snapshot's table-property
+	// configuration. Returns NONE when the property is absent or `"none"`.
+	static DeltaColumnMappingMode ReadColumnMappingMode(ffi::SharedSnapshot *snapshot);
 
 	static void *StringAllocationNew(const struct ffi::KernelStringSlice slice) {
 		return new string(slice.ptr, slice.len);
@@ -276,12 +286,15 @@ struct DeltaMultiFileColumnDefinition : public MultiFileColumnDefinition {
 // SchemaVisitor is used to parse the schema of a Delta table from the Kernel
 class SchemaVisitor {
 public:
-	explicit SchemaVisitor(ffi::Handle<ffi::SharedExternEngine> engine_p) : engine(engine_p) {};
+	SchemaVisitor(ffi::Handle<ffi::SharedExternEngine> engine_p, DeltaColumnMappingMode mapping_mode_p)
+	    : engine(engine_p), mapping_mode(mapping_mode_p) {};
 
 	static vector<DeltaMultiFileColumnDefinition> VisitSnapshotSchema(ffi::Handle<ffi::SharedExternEngine> engine,
-	                                                                  ffi::SharedSnapshot *snapshot);
+	                                                                  ffi::SharedSnapshot *snapshot,
+	                                                                  DeltaColumnMappingMode mapping_mode);
 	static vector<DeltaMultiFileColumnDefinition>
-	VisitSnapshotGlobalReadSchema(ffi::Handle<ffi::SharedExternEngine> engine, ffi::SharedScan *state, bool logical);
+	VisitSnapshotGlobalReadSchema(ffi::Handle<ffi::SharedExternEngine> engine, ffi::SharedScan *state, bool logical,
+	                              DeltaColumnMappingMode mapping_mode);
 	static vector<DeltaMultiFileColumnDefinition> VisitWriteContextSchema(ffi::Handle<ffi::SharedExternEngine> engine,
 	                                                                      ffi::SharedWriteContext *write_context);
 
@@ -290,22 +303,38 @@ class SchemaVisitor {
 	uintptr_t next_id = 1;
 
 	ffi::SharedExternEngine *engine = nullptr;
+	DeltaColumnMappingMode mapping_mode = DeltaColumnMappingMode::NONE;
 	ErrorData error;
 
 	static ffi::EngineSchemaVisitor CreateSchemaVisitor(SchemaVisitor &state);
 
 	typedef void(SimpleTypeVisitorFunction)(void *, uintptr_t, ffi::KernelStringSlice, bool is_nullable,
 	                                        const ffi::CStringMap *metadata);
 
-	static void ApplyDeltaColumnMapping(ffi::Handle<ffi::SharedExternEngine> engine, const ffi::CStringMap *metadata,
+	// Set `col_def.identifier` so DuckDB's MultiFileReader resolves the column
+	// the way the Delta protocol's "Reader Requirements for Column Mapping"
+	// require for the active mode. Identifier type drives the dispatch:
+	// BIGINT -> match by parquet field_id, VARCHAR -> match by name. Leaving
+	// it unset matches by display (logical) name.
+	static void ApplyDeltaColumnMapping(SchemaVisitor &state, const ffi::CStringMap *metadata,
 	                                    DeltaMultiFileColumnDefinition &col_def) {
-		auto id = KernelUtils::FetchFromStringMap(engine, metadata, "parquet.field.id");
-		if (!id.empty()) {
-			col_def.identifier = Value(id).DefaultCastAs(LogicalType::BIGINT);
+		switch (state.mapping_mode) {
+		case DeltaColumnMappingMode::ID: {
+			auto id = KernelUtils::FetchFromStringMap(state.engine, metadata, "parquet.field.id");
+			if (!id.empty()) {
+				col_def.identifier = Value(id).DefaultCastAs(LogicalType::BIGINT);
+			}
+			break;
+		}
+		case DeltaColumnMappingMode::NAME: {
+			auto name = KernelUtils::FetchFromStringMap(state.engine, metadata, "delta.columnMapping.physicalName");
+			if (!name.empty()) {
+				col_def.identifier = Value(name);
+			}
+			break;
 		}
-		auto name = KernelUtils::FetchFromStringMap(engine, metadata, "delta.columnMapping.physicalName");
-		if (!name.empty()) {
-			col_def.identifier = Value(name);
+		case DeltaColumnMappingMode::NONE:
+			break;
 		}
 		col_def.default_expression = make_uniq<ConstantExpression>(Value(col_def.type));
 	}
@@ -318,7 +347,7 @@ class SchemaVisitor {
 	static void VisitSimpleTypeImpl(SchemaVisitor *state, uintptr_t sibling_list_id, ffi::KernelStringSlice name,
 	                                bool is_nullable, const ffi::CStringMap *metadata) {
 		DeltaMultiFileColumnDefinition col_def(KernelUtils::FromDeltaString(name), TypeId, is_nullable);
-		ApplyDeltaColumnMapping(state->engine, metadata, col_def);
+		ApplyDeltaColumnMapping(*state, metadata, col_def);
 
 		state->AppendToList(sibling_list_id, name, std::move(col_def));
 	}

src/delta_utils.cpp

@@ -0,0 +1,22 @@
+# name: test/sql/inlined/column_mapping_id_mode_logical_names.test
+# description: id-mode tables whose parquet files carry logical column names plus
+#              PARQUET:field_id metadata must resolve by field_id, not by name.
+# group: [inlined]
+
+require parquet
+
+require delta
+
+statement ok
+from copy_dir('data/inlined/column_mapping_id_mode_logical_names',
+              '__TEST_DIR__/data/inlined/column_mapping_id_mode_logical_names');
+
+# Bug repro: the parquet file's columns are named "a"/"b" (the logical names)
+# and carry field_ids 1/2. The Delta log declares id-mode column mapping with
+# physical names "col-a-uuid"/"col-b-uuid". Resolving by physical name produces
+# all-NULL rows. Resolving by field_id (per the protocol) returns the data.
+query II
+from delta_scan('__TEST_DIR__/data/inlined/column_mapping_id_mode_logical_names/delta_lake') order by a
+----
+1	alice
+2	bob