Skip to content

Commit f3359c7

Browse files
committed
create v2 adapters for metadata and protocol
1 parent ecf4948 commit f3359c7

5 files changed

Lines changed: 421 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

Lines changed: 103 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,103 @@
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.List;
24+
import java.util.Objects;
25+
import org.apache.spark.sql.delta.DeltaColumnMappingMode;
26+
import org.apache.spark.sql.delta.DeltaColumnMappingMode$;
27+
import org.apache.spark.sql.delta.NoMapping$;
28+
import org.apache.spark.sql.delta.v2.interop.AbstractMetadata;
29+
import org.apache.spark.sql.types.StructType;
30+
import scala.collection.immutable.Map;
31+
import scala.collection.immutable.Seq;
32+
import scala.jdk.javaapi.CollectionConverters;
33+
34+
/**
35+
* Adapter from {@link io.delta.kernel.internal.actions.Metadata} to {@link
36+
* org.apache.spark.sql.delta.v2.interop.AbstractMetadata}.
37+
*/
38+
public class KernelMetadataAdapter implements AbstractMetadata {
39+
40+
private final Metadata kernelMetadata;
41+
private volatile StructType cachedSchema;
42+
private volatile Seq<String> cachedPartitionColumns;
43+
private volatile Map<String, String> cachedConfiguration;
44+
private volatile StructType cachedPartitionSchema;
45+
46+
public KernelMetadataAdapter(Metadata kernelMetadata) {
47+
this.kernelMetadata = Objects.requireNonNull(kernelMetadata, "kernelMetadata is null");
48+
}
49+
50+
@Override
51+
public String id() {
52+
return kernelMetadata.getId();
53+
}
54+
55+
@Override
56+
public String name() {
57+
return kernelMetadata.getName().orElse(null);
58+
}
59+
60+
@Override
61+
public String description() {
62+
return kernelMetadata.getDescription().orElse(null);
63+
}
64+
65+
@Override
66+
public StructType schema() {
67+
if (cachedSchema == null) {
68+
cachedSchema = SchemaUtils.convertKernelSchemaToSparkSchema(kernelMetadata.getSchema());
69+
}
70+
return cachedSchema;
71+
}
72+
73+
@Override
74+
public Seq<String> partitionColumns() {
75+
if (cachedPartitionColumns == null) {
76+
List<String> rawCols = VectorUtils.toJavaList(kernelMetadata.getPartitionColumns());
77+
cachedPartitionColumns = CollectionConverters.asScala(rawCols).toSeq();
78+
}
79+
return cachedPartitionColumns;
80+
}
81+
82+
@Override
83+
public Map<String, String> configuration() {
84+
if (cachedConfiguration == null) {
85+
cachedConfiguration = ScalaUtils.toScalaMap(kernelMetadata.getConfiguration());
86+
}
87+
return cachedConfiguration;
88+
}
89+
90+
@Override
91+
public DeltaColumnMappingMode columnMappingMode() {
92+
String mode = kernelMetadata.getConfiguration().get(ColumnMapping.COLUMN_MAPPING_MODE_KEY);
93+
return mode == null ? NoMapping$.MODULE$ : DeltaColumnMappingMode$.MODULE$.apply(mode);
94+
}
95+
96+
@Override
97+
public StructType partitionSchema() {
98+
if (cachedPartitionSchema == null) {
99+
cachedPartitionSchema = AbstractMetadata.super.partitionSchema();
100+
}
101+
return cachedPartitionSchema;
102+
}
103+
}
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+
}

0 commit comments

Comments
 (0)