Skip to content

Commit dfe61d9

Browse files
PorridgeSwimPorridgeSwim
committed
create v2 adapters for metadata and protocol
1 parent 9436fca commit dfe61d9

5 files changed

Lines changed: 365 additions & 2 deletions

File tree

‎spark/src/main/scala/org/apache/spark/sql/delta/actions/actions.scala‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1253,8 +1253,7 @@ case class Metadata(
12531253

12541254
/** Returns the partitionSchema as a [[StructType]] */
12551255
@JsonIgnore
1256-
lazy val partitionSchema: StructType =
1257-
new StructType(partitionColumns.map(c => schema(c)).toArray)
1256+
override lazy val partitionSchema: StructType = super.partitionSchema
12581257

12591258
/** Partition value keys in the AddFile map. */
12601259
@JsonIgnore

‎spark/src/main/scala/org/apache/spark/sql/delta/v2/interop/AbstractMetadata.scala‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616

1717
package org.apache.spark.sql.delta.v2.interop
1818

19+
import org.apache.spark.sql.delta.DeltaColumnMappingMode
1920
import org.apache.spark.sql.types.StructType
2021

2122
/**
@@ -42,5 +43,12 @@ trait AbstractMetadata {
4243

4344
/** The table properties/configuration defined on the table. */
4445
def configuration: Map[String, String]
46+
47+
/** Column mapping mode for this table. */
48+
def columnMappingMode: DeltaColumnMappingMode
49+
50+
/** Returns the partitionSchema as a [[StructType]] */
51+
def partitionSchema: StructType =
52+
new StructType(partitionColumns.map(c => schema(c)).toArray)
4553
}
4654

‎spark/v2/src/main/java/io/delta/spark/internal/v2/adapters/KernelMetadataAdapter.java‎

Lines changed: 118 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,118 @@
1+
/*
2+
* Copyright (2026) The Delta Lake Project Authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package io.delta.spark.internal.v2.adapters;
17+
18+
import io.delta.kernel.internal.actions.Metadata;
19+
import io.delta.kernel.internal.util.ColumnMapping;
20+
import io.delta.kernel.internal.util.VectorUtils;
21+
import io.delta.spark.internal.v2.utils.ScalaUtils;
22+
import io.delta.spark.internal.v2.utils.SchemaUtils;
23+
import java.util.Objects;
24+
import java.util.stream.Collectors;
25+
import org.apache.spark.sql.delta.DeltaColumnMappingMode;
26+
import org.apache.spark.sql.delta.IdMapping$;
27+
import org.apache.spark.sql.delta.NameMapping$;
28+
import org.apache.spark.sql.delta.NoMapping$;
29+
import org.apache.spark.sql.delta.v2.interop.AbstractMetadata;
30+
import org.apache.spark.sql.types.StructType;
31+
import scala.collection.immutable.Map;
32+
import scala.collection.immutable.Seq;
33+
import scala.jdk.javaapi.CollectionConverters;
34+
35+
/**
36+
* Adapter from {@link io.delta.kernel.internal.actions.Metadata} to {@link
37+
* org.apache.spark.sql.delta.v2.interop.AbstractMetadata}.
38+
*/
39+
public class KernelMetadataAdapter implements AbstractMetadata {
40+
41+
private final Metadata kernelMetadata;
42+
private volatile StructType cachedSchema;
43+
private volatile Seq<String> cachedPartitionColumns;
44+
private volatile Map<String, String> cachedConfiguration;
45+
private volatile StructType cachedPartitionSchema;
46+
47+
public KernelMetadataAdapter(Metadata kernelMetadata) {
48+
this.kernelMetadata = Objects.requireNonNull(kernelMetadata, "kernelMetadata is null");
49+
}
50+
51+
@Override
52+
public String id() {
53+
return kernelMetadata.getId();
54+
}
55+
56+
@Override
57+
public String name() {
58+
return kernelMetadata.getName().orElse(null);
59+
}
60+
61+
@Override
62+
public String description() {
63+
return kernelMetadata.getDescription().orElse(null);
64+
}
65+
66+
@Override
67+
public StructType schema() {
68+
if (cachedSchema == null) {
69+
cachedSchema = SchemaUtils.convertKernelSchemaToSparkSchema(kernelMetadata.getSchema());
70+
}
71+
return cachedSchema;
72+
}
73+
74+
@Override
75+
public Seq<String> partitionColumns() {
76+
if (cachedPartitionColumns == null) {
77+
cachedPartitionColumns =
78+
CollectionConverters.asScala(
79+
VectorUtils.toJavaList(kernelMetadata.getPartitionColumns()).stream()
80+
.map(Object::toString)
81+
.collect(Collectors.toList()))
82+
.toSeq();
83+
}
84+
return cachedPartitionColumns;
85+
}
86+
87+
@Override
88+
public Map<String, String> configuration() {
89+
if (cachedConfiguration == null) {
90+
cachedConfiguration = ScalaUtils.toScalaMap(kernelMetadata.getConfiguration());
91+
}
92+
return cachedConfiguration;
93+
}
94+
95+
@Override
96+
public DeltaColumnMappingMode columnMappingMode() {
97+
ColumnMapping.ColumnMappingMode kernelMode =
98+
ColumnMapping.getColumnMappingMode(kernelMetadata.getConfiguration());
99+
switch (kernelMode) {
100+
case NONE:
101+
return NoMapping$.MODULE$;
102+
case ID:
103+
return IdMapping$.MODULE$;
104+
case NAME:
105+
return NameMapping$.MODULE$;
106+
default:
107+
throw new UnsupportedOperationException("Unsupported column mapping mode: " + kernelMode);
108+
}
109+
}
110+
111+
@Override
112+
public StructType partitionSchema() {
113+
if (cachedPartitionSchema == null) {
114+
cachedPartitionSchema = AbstractMetadata.super.partitionSchema();
115+
}
116+
return cachedPartitionSchema;
117+
}
118+
}

‎spark/v2/src/main/java/io/delta/spark/internal/v2/adapters/KernelProtocolAdapter.java‎

Lines changed: 72 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,72 @@
1+
/*
2+
* Copyright (2026) The Delta Lake Project Authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package io.delta.spark.internal.v2.adapters;
17+
18+
import io.delta.kernel.internal.actions.Protocol;
19+
import java.util.Objects;
20+
import org.apache.spark.sql.delta.v2.interop.AbstractProtocol;
21+
import scala.Option;
22+
import scala.collection.immutable.Set;
23+
import scala.jdk.javaapi.CollectionConverters;
24+
25+
/**
26+
* Adapter from {@link io.delta.kernel.internal.actions.Protocol} to {@link
27+
* org.apache.spark.sql.delta.v2.interop.AbstractProtocol}.
28+
*/
29+
public class KernelProtocolAdapter implements AbstractProtocol {
30+
31+
private final Protocol kernelProtocol;
32+
private volatile Option<Set<String>> cachedReaderFeatures;
33+
private volatile Option<Set<String>> cachedWriterFeatures;
34+
35+
public KernelProtocolAdapter(Protocol kernelProtocol) {
36+
this.kernelProtocol = Objects.requireNonNull(kernelProtocol, "kernelProtocol is null");
37+
}
38+
39+
@Override
40+
public int minReaderVersion() {
41+
return kernelProtocol.getMinReaderVersion();
42+
}
43+
44+
@Override
45+
public int minWriterVersion() {
46+
return kernelProtocol.getMinWriterVersion();
47+
}
48+
49+
@Override
50+
public Option<Set<String>> readerFeatures() {
51+
if (cachedReaderFeatures == null) {
52+
cachedReaderFeatures =
53+
kernelProtocol.supportsReaderFeatures()
54+
? Option.apply(
55+
CollectionConverters.asScala(kernelProtocol.getReaderFeatures()).toSet())
56+
: Option.empty();
57+
}
58+
return cachedReaderFeatures;
59+
}
60+
61+
@Override
62+
public Option<Set<String>> writerFeatures() {
63+
if (cachedWriterFeatures == null) {
64+
cachedWriterFeatures =
65+
kernelProtocol.supportsWriterFeatures()
66+
? Option.apply(
67+
CollectionConverters.asScala(kernelProtocol.getWriterFeatures()).toSet())
68+
: Option.empty();
69+
}
70+
return cachedWriterFeatures;
71+
}
72+
}

‎spark/v2/src/test/java/io/delta/spark/internal/v2/adapters/ActionAdaptersTest.java‎

Lines changed: 166 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,166 @@
1+
/*
2+
* Copyright (2026) The Delta Lake Project Authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package io.delta.spark.internal.v2.adapters;
17+
18+
import static org.junit.jupiter.api.Assertions.*;
19+
20+
import io.delta.kernel.data.ArrayValue;
21+
import io.delta.kernel.internal.actions.Format;
22+
import io.delta.kernel.internal.actions.Metadata;
23+
import io.delta.kernel.internal.actions.Protocol;
24+
import io.delta.kernel.internal.util.VectorUtils;
25+
import io.delta.kernel.types.IntegerType;
26+
import io.delta.kernel.types.StringType;
27+
import io.delta.kernel.types.StructType;
28+
import java.util.*;
29+
import org.apache.spark.sql.delta.NameMapping$;
30+
import org.apache.spark.sql.delta.NoMapping$;
31+
import org.junit.jupiter.api.Test;
32+
import scala.jdk.javaapi.CollectionConverters;
33+
34+
/** Unit tests for {@link KernelMetadataAdapter} and {@link KernelProtocolAdapter}. */
35+
public class ActionAdaptersTest {
36+
37+
// ===== KernelProtocolAdapter =====
38+
39+
@Test
40+
public void testProtocolAdapterWithTableFeatures() {
41+
// Reader features: supported but empty (version >= 3 means features are supported, even with
42+
// an empty set). Writer features: supported and populated.
43+
Set<String> readerFeatures = Collections.emptySet();
44+
Set<String> writerFeatures = new HashSet<>(Arrays.asList("v2Checkpoint", "rowTracking"));
45+
Protocol kernelProtocol = new Protocol(3, 7, readerFeatures, writerFeatures);
46+
47+
KernelProtocolAdapter adapter = new KernelProtocolAdapter(kernelProtocol);
48+
49+
assertEquals(3, adapter.minReaderVersion());
50+
assertEquals(7, adapter.minWriterVersion());
51+
assertTrue(adapter.readerFeatures().isDefined());
52+
assertTrue(CollectionConverters.asJava(adapter.readerFeatures().get()).isEmpty());
53+
assertTrue(adapter.writerFeatures().isDefined());
54+
assertEquals(
55+
new HashSet<>(Arrays.asList("v2Checkpoint", "rowTracking")),
56+
CollectionConverters.asJava(adapter.writerFeatures().get()));
57+
}
58+
59+
@Test
60+
public void testProtocolAdapterLegacyProtocol() {
61+
Protocol kernelProtocol = new Protocol(1, 2);
62+
63+
KernelProtocolAdapter adapter = new KernelProtocolAdapter(kernelProtocol);
64+
65+
assertEquals(1, adapter.minReaderVersion());
66+
assertEquals(2, adapter.minWriterVersion());
67+
assertTrue(adapter.readerFeatures().isEmpty());
68+
assertTrue(adapter.writerFeatures().isEmpty());
69+
}
70+
71+
@Test
72+
public void testProtocolAdapterNullThrows() {
73+
assertThrows(NullPointerException.class, () -> new KernelProtocolAdapter(null));
74+
}
75+
76+
// ===== KernelMetadataAdapter =====
77+
78+
@Test
79+
public void testMetadataAdapter() {
80+
ArrayValue partCols =
81+
VectorUtils.buildArrayValue(Arrays.asList("part1", "part2"), StringType.STRING);
82+
Map<String, String> formatOptions = Collections.singletonMap("foo", "bar");
83+
Format format = new Format("parquet", formatOptions);
84+
Map<String, String> configuration = new HashMap<>();
85+
configuration.put("zip", "zap");
86+
configuration.put("delta.columnMapping.mode", "name");
87+
88+
Metadata kernelMetadata =
89+
new Metadata(
90+
"id",
91+
Optional.of("name"),
92+
Optional.of("description"),
93+
format,
94+
"{\"type\":\"struct\",\"fields\":"
95+
+ "[{\"name\":\"part1\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},"
96+
+ "{\"name\":\"part2\",\"type\":\"string\",\"nullable\":false,\"metadata\":{}},"
97+
+ "{\"name\":\"col1\",\"type\":\"string\",\"nullable\":false,\"metadata\":{}}]}",
98+
new StructType()
99+
.add("part1", IntegerType.INTEGER)
100+
.add("part2", StringType.STRING, false /* nullable */)
101+
.add("col1", StringType.STRING, false /* nullable */),
102+
partCols,
103+
Optional.of(42L),
104+
VectorUtils.stringStringMapValue(configuration));
105+
106+
KernelMetadataAdapter adapter = new KernelMetadataAdapter(kernelMetadata);
107+
108+
assertEquals("id", adapter.id());
109+
assertEquals("name", adapter.name());
110+
assertEquals("description", adapter.description());
111+
assertEquals(3, adapter.schema().fields().length);
112+
assertEquals("integer", adapter.schema().apply("part1").dataType().typeName());
113+
assertTrue(adapter.schema().apply("part1").nullable());
114+
assertEquals("string", adapter.schema().apply("part2").dataType().typeName());
115+
assertFalse(adapter.schema().apply("part2").nullable());
116+
assertEquals("string", adapter.schema().apply("col1").dataType().typeName());
117+
assertFalse(adapter.schema().apply("col1").nullable());
118+
assertEquals(
119+
Arrays.asList("part1", "part2"), CollectionConverters.asJava(adapter.partitionColumns()));
120+
org.apache.spark.sql.types.StructType partSchema = adapter.partitionSchema();
121+
assertEquals(2, partSchema.fields().length);
122+
assertEquals("part1", partSchema.fields()[0].name());
123+
assertEquals("integer", partSchema.fields()[0].dataType().typeName());
124+
assertTrue(partSchema.fields()[0].nullable());
125+
assertEquals("part2", partSchema.fields()[1].name());
126+
assertEquals("string", partSchema.fields()[1].dataType().typeName());
127+
assertFalse(partSchema.fields()[1].nullable());
128+
assertEquals(configuration, CollectionConverters.asJava(adapter.configuration()));
129+
assertEquals(NameMapping$.MODULE$, adapter.columnMappingMode());
130+
}
131+
132+
@Test
133+
public void testMetadataAdapterWithNullOptionalFields() {
134+
ArrayValue emptyPartCols =
135+
VectorUtils.buildArrayValue(Collections.emptyList(), StringType.STRING);
136+
Format format = new Format("parquet", Collections.emptyMap());
137+
138+
Metadata kernelMetadata =
139+
new Metadata(
140+
"id2",
141+
Optional.empty(),
142+
Optional.empty(),
143+
format,
144+
"{\"type\":\"struct\",\"fields\":[]}",
145+
new StructType(),
146+
emptyPartCols,
147+
Optional.empty(),
148+
VectorUtils.stringStringMapValue(Collections.emptyMap()));
149+
150+
KernelMetadataAdapter adapter = new KernelMetadataAdapter(kernelMetadata);
151+
152+
assertEquals("id2", adapter.id());
153+
assertNull(adapter.name());
154+
assertNull(adapter.description());
155+
assertEquals(0, adapter.schema().fields().length);
156+
assertTrue(CollectionConverters.asJava(adapter.partitionColumns()).isEmpty());
157+
assertEquals(0, adapter.partitionSchema().fields().length);
158+
assertTrue(CollectionConverters.asJava(adapter.configuration()).isEmpty());
159+
assertEquals(NoMapping$.MODULE$, adapter.columnMappingMode());
160+
}
161+
162+
@Test
163+
public void testMetadataAdapterNullThrows() {
164+
assertThrows(NullPointerException.class, () -> new KernelMetadataAdapter(null));
165+
}
166+
}

0 commit comments

Comments
 (0)