This is an automated email from the ASF dual-hosted git repository.
vinishjail97 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-xtable.git
The following commit(s) were added to refs/heads/main by this push:
new dd15df94 [911] Carry Delta's nested field ids into the converted
schema (#912)
dd15df94 is described below
commit dd15df94f8ff3f69638a72e4815c04d78c01f3c7
Author: Sylwester Lachiewicz <[email protected]>
AuthorDate: Tue Sep 15 01:41:09 2026 +0200
[911] Carry Delta's nested field ids into the converted schema (#912)
* [911] Carry Delta's nested field ids into the converted schema
Delta assigns column mapping ids to struct fields only, so the key, value
and element of a collection have no id of their own. When IcebergCompatV2
is enabled Delta does assign them and records them per field under
delta.columnMapping.nested.ids, keyed by the path the child takes in the
parquet file. Read those and put them on the converted fields, which the
Iceberg target already prefers over the ids it generates itself.
* [911] Carry Delta's nested field ids into the converted schema with
complex test cases and kernel fix
* [911] Move the shared Delta metadata keys into
org.apache.xtable.delta.Constants
The Spark based and the Kernel based schema extractors each declared their
own copy of the Delta column mapping keys and of the parquet child names
used
in delta.columnMapping.nested.ids. delta.generationExpression was declared
in
three classes, and the copy in DeltaKernelSchemaExtractor was unused.
Collect them in a package level Constants class, following the style of
org.apache.xtable.catalog.Constants, and reference them by static import.
---------
Co-authored-by: Vinish Reddy <[email protected]>
---
.gitignore | 1 +
.../java/org/apache/xtable/delta/Constants.java | 57 +++++++
.../xtable/delta/DeltaPartitionExtractor.java | 2 +-
.../apache/xtable/delta/DeltaSchemaExtractor.java | 70 ++++++--
.../kernel/DeltaKernelPartitionExtractor.java | 2 +-
.../xtable/kernel/DeltaKernelSchemaExtractor.java | 62 ++++++--
.../xtable/delta/TestDeltaPartitionExtractor.java | 2 +-
.../xtable/delta/TestDeltaSchemaExtractor.java | 176 +++++++++++++++++++++
.../kernel/TestDeltaKernelPartitionExtractor.java | 2 +-
.../kernel/TestDeltaKernelSchemaExtractor.java | 127 +++++++++++++++
10 files changed, 471 insertions(+), 30 deletions(-)
diff --git a/.gitignore b/.gitignore
index 4273535c..def30ad3 100644
--- a/.gitignore
+++ b/.gitignore
@@ -27,6 +27,7 @@ hs_err_pid*
.java-version
.idea
.vscode
+.claude/
# Ignore Gradle project-specific cache directory
.gradle
diff --git a/xtable-core/src/main/java/org/apache/xtable/delta/Constants.java
b/xtable-core/src/main/java/org/apache/xtable/delta/Constants.java
new file mode 100644
index 00000000..ccaeae1a
--- /dev/null
+++ b/xtable-core/src/main/java/org/apache/xtable/delta/Constants.java
@@ -0,0 +1,57 @@
+/*
+ * 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.
+ */
+
+package org.apache.xtable.delta;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+/**
+ * Field metadata keys that Delta Lake writes into a table's schema, and the
names Delta uses for
+ * the children of a collection within them. The Spark based and the Kernel
based readers both read
+ * the same keys, so they share these definitions.
+ */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class Constants {
+
+ /** The column mapping id Delta assigns to a struct field, which is also its
parquet field id. */
+ public static final String DELTA_COLUMN_MAPPING_ID =
"delta.columnMapping.id";
+
+ /** The physical name a struct field takes in the parquet file under column
mapping. */
+ public static final String DELTA_COLUMN_MAPPING_NAME =
"delta.columnMapping.physicalName";
+
+ /**
+ * Written by Delta 3.x when IcebergCompatV2 is enabled. Holds the parquet
field ids of a field's
+ * map keys, map values and list elements, keyed by the path those take in
the file, for example
+ * "col-1234.key". Delta assigns column mapping ids to struct fields only,
so without these the
+ * nested fields of a collection have no id to match a reader against.
+ */
+ public static final String DELTA_COLUMN_MAPPING_NESTED_IDS =
"delta.columnMapping.nested.ids";
+
+ /** The expression that generates the values of a Delta generated column. */
+ public static final String DELTA_GENERATION_EXPRESSION =
"delta.generationExpression";
+
+ /** The name parquet gives a list element, as used in {@link
#DELTA_COLUMN_MAPPING_NESTED_IDS}. */
+ public static final String PARQUET_LIST_ELEMENT_FIELD_NAME = "element";
+
+ /** The name parquet gives a map key, as used in {@link
#DELTA_COLUMN_MAPPING_NESTED_IDS}. */
+ public static final String PARQUET_MAP_KEY_FIELD_NAME = "key";
+
+ /** The name parquet gives a map value, as used in {@link
#DELTA_COLUMN_MAPPING_NESTED_IDS}. */
+ public static final String PARQUET_MAP_VALUE_FIELD_NAME = "value";
+}
diff --git
a/xtable-core/src/main/java/org/apache/xtable/delta/DeltaPartitionExtractor.java
b/xtable-core/src/main/java/org/apache/xtable/delta/DeltaPartitionExtractor.java
index 326b03c0..62aaa8cc 100644
---
a/xtable-core/src/main/java/org/apache/xtable/delta/DeltaPartitionExtractor.java
+++
b/xtable-core/src/main/java/org/apache/xtable/delta/DeltaPartitionExtractor.java
@@ -19,6 +19,7 @@
package org.apache.xtable.delta;
import static org.apache.xtable.collectors.CustomCollectors.toList;
+import static org.apache.xtable.delta.Constants.DELTA_GENERATION_EXPRESSION;
import static
org.apache.xtable.delta.DeltaValueConverter.convertFromDeltaPartitionValue;
import static
org.apache.xtable.delta.DeltaValueConverter.convertToDeltaPartitionValue;
@@ -79,7 +80,6 @@ public class DeltaPartitionExtractor {
// For timestamp partition fields, actual partition column names in delta
format will be of type
// generated & and with a name like
`delta_partition_col_{transform_type}_{source_field_name}`.
private static final String DELTA_PARTITION_COL_NAME_FORMAT =
"xtable_partition_col_%s_%s";
- public static final String DELTA_GENERATION_EXPRESSION =
"delta.generationExpression";
private static final List<ParsedGeneratedExpr.GeneratedExprType>
GRANULARITIES =
Arrays.asList(
ParsedGeneratedExpr.GeneratedExprType.YEAR,
diff --git
a/xtable-core/src/main/java/org/apache/xtable/delta/DeltaSchemaExtractor.java
b/xtable-core/src/main/java/org/apache/xtable/delta/DeltaSchemaExtractor.java
index f72d5a6a..ed10f915 100644
---
a/xtable-core/src/main/java/org/apache/xtable/delta/DeltaSchemaExtractor.java
+++
b/xtable-core/src/main/java/org/apache/xtable/delta/DeltaSchemaExtractor.java
@@ -18,6 +18,14 @@
package org.apache.xtable.delta;
+import static org.apache.xtable.delta.Constants.DELTA_COLUMN_MAPPING_ID;
+import static org.apache.xtable.delta.Constants.DELTA_COLUMN_MAPPING_NAME;
+import static
org.apache.xtable.delta.Constants.DELTA_COLUMN_MAPPING_NESTED_IDS;
+import static org.apache.xtable.delta.Constants.DELTA_GENERATION_EXPRESSION;
+import static
org.apache.xtable.delta.Constants.PARQUET_LIST_ELEMENT_FIELD_NAME;
+import static org.apache.xtable.delta.Constants.PARQUET_MAP_KEY_FIELD_NAME;
+import static org.apache.xtable.delta.Constants.PARQUET_MAP_VALUE_FIELD_NAME;
+
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
@@ -54,8 +62,6 @@ import org.apache.xtable.schema.SchemaUtils;
*/
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public class DeltaSchemaExtractor {
- private static final String DELTA_COLUMN_MAPPING_ID =
"delta.columnMapping.id";
- private static final String DELTA_COLUMN_MAPPING_NAME =
"delta.columnMapping.physicalName";
private static final DeltaSchemaExtractor INSTANCE = new
DeltaSchemaExtractor();
// Timestamps in Delta are microsecond precision by default
private static final Map<InternalSchema.MetadataKey, Object>
@@ -67,8 +73,19 @@ public class DeltaSchemaExtractor {
return INSTANCE;
}
+ private static String childIdPath(String nestedIdPath, String childName) {
+ return nestedIdPath == null ? null : nestedIdPath + "." + childName;
+ }
+
+ private static Integer nestedFieldId(Metadata nestedIds, String path) {
+ if (nestedIds == null || path == null || !nestedIds.contains(path)) {
+ return null;
+ }
+ return (int) nestedIds.getLong(path);
+ }
+
public InternalSchema toInternalSchema(StructType structType) {
- return toInternalSchema(structType, null, false, null, null);
+ return toInternalSchema(structType, null, false, null, null, null, null);
}
private InternalSchema toInternalSchema(
@@ -77,6 +94,23 @@ public class DeltaSchemaExtractor {
boolean nullable,
String comment,
Metadata originalMetadata) {
+ return toInternalSchema(dataType, parentPath, nullable, comment,
originalMetadata, null, null);
+ }
+
+ /**
+ * @param nestedIds the {@code delta.columnMapping.nested.ids} metadata of
the enclosing struct
+ * field, or null when the field carries none
+ * @param nestedIdPath the path of the current type within that metadata,
starting at the
+ * enclosing field's physical name
+ */
+ private InternalSchema toInternalSchema(
+ DataType dataType,
+ String parentPath,
+ boolean nullable,
+ String comment,
+ Metadata originalMetadata,
+ Metadata nestedIds,
+ String nestedIdPath) {
Map<InternalSchema.MetadataKey, Object> metadata = null;
List<InternalField> fields = null;
InternalType type;
@@ -127,11 +161,7 @@ public class DeltaSchemaExtractor {
StructType structType = (StructType) dataType;
fields =
Arrays.stream(structType.fields())
- .filter(
- field ->
- !field
- .metadata()
-
.contains(DeltaPartitionExtractor.DELTA_GENERATION_EXPRESSION))
+ .filter(field ->
!field.metadata().contains(DELTA_GENERATION_EXPRESSION))
.map(
field -> {
Integer fieldId =
@@ -150,7 +180,11 @@ public class DeltaSchemaExtractor {
SchemaUtils.getFullyQualifiedPath(parentPath,
field.name()),
field.nullable(),
fieldComment,
- field.metadata());
+ field.metadata(),
+
field.metadata().contains(DELTA_COLUMN_MAPPING_NESTED_IDS)
+ ?
field.metadata().getMetadata(DELTA_COLUMN_MAPPING_NESTED_IDS)
+ : null,
+ storageName != null ? storageName :
field.name());
return InternalField.builder()
.name(field.name())
.fieldId(fieldId)
@@ -173,6 +207,7 @@ public class DeltaSchemaExtractor {
break;
case "array":
ArrayType arrayType = (ArrayType) dataType;
+ String elementIdPath = childIdPath(nestedIdPath,
PARQUET_LIST_ELEMENT_FIELD_NAME);
InternalSchema elementSchema =
toInternalSchema(
arrayType.elementType(),
@@ -180,11 +215,14 @@ public class DeltaSchemaExtractor {
parentPath,
InternalField.Constants.ARRAY_ELEMENT_FIELD_NAME),
arrayType.containsNull(),
null,
- null);
+ null,
+ nestedIds,
+ elementIdPath);
InternalField elementField =
InternalField.builder()
.name(InternalField.Constants.ARRAY_ELEMENT_FIELD_NAME)
.parentPath(parentPath)
+ .fieldId(nestedFieldId(nestedIds, elementIdPath))
.schema(elementSchema)
.build();
type = InternalType.LIST;
@@ -192,6 +230,8 @@ public class DeltaSchemaExtractor {
break;
case "map":
MapType mapType = (MapType) dataType;
+ String keyIdPath = childIdPath(nestedIdPath,
PARQUET_MAP_KEY_FIELD_NAME);
+ String valueIdPath = childIdPath(nestedIdPath,
PARQUET_MAP_VALUE_FIELD_NAME);
InternalSchema keySchema =
toInternalSchema(
mapType.keyType(),
@@ -199,11 +239,14 @@ public class DeltaSchemaExtractor {
parentPath, InternalField.Constants.MAP_KEY_FIELD_NAME),
false,
null,
- null);
+ null,
+ nestedIds,
+ keyIdPath);
InternalField keyField =
InternalField.builder()
.name(InternalField.Constants.MAP_KEY_FIELD_NAME)
.parentPath(parentPath)
+ .fieldId(nestedFieldId(nestedIds, keyIdPath))
.schema(keySchema)
.build();
InternalSchema valueSchema =
@@ -213,11 +256,14 @@ public class DeltaSchemaExtractor {
parentPath, InternalField.Constants.MAP_VALUE_FIELD_NAME),
mapType.valueContainsNull(),
null,
- null);
+ null,
+ nestedIds,
+ valueIdPath);
InternalField valueField =
InternalField.builder()
.name(InternalField.Constants.MAP_VALUE_FIELD_NAME)
.parentPath(parentPath)
+ .fieldId(nestedFieldId(nestedIds, valueIdPath))
.schema(valueSchema)
.build();
type = InternalType.MAP;
diff --git
a/xtable-core/src/main/java/org/apache/xtable/kernel/DeltaKernelPartitionExtractor.java
b/xtable-core/src/main/java/org/apache/xtable/kernel/DeltaKernelPartitionExtractor.java
index 4a236867..01532041 100644
---
a/xtable-core/src/main/java/org/apache/xtable/kernel/DeltaKernelPartitionExtractor.java
+++
b/xtable-core/src/main/java/org/apache/xtable/kernel/DeltaKernelPartitionExtractor.java
@@ -19,6 +19,7 @@
package org.apache.xtable.kernel;
import static org.apache.xtable.collectors.CustomCollectors.toList;
+import static org.apache.xtable.delta.Constants.DELTA_GENERATION_EXPRESSION;
import static
org.apache.xtable.delta.DeltaValueConverter.convertFromDeltaPartitionValue;
import static
org.apache.xtable.delta.DeltaValueConverter.convertToDeltaPartitionValue;
@@ -73,7 +74,6 @@ public class DeltaKernelPartitionExtractor {
// For timestamp partition fields, actual partition column names in delta
format will be of type
// generated & and with a name like
`delta_partition_col_{transform_type}_{source_field_name}`.
private static final String DELTA_PARTITION_COL_NAME_FORMAT =
"xtable_partition_col_%s_%s";
- static final String DELTA_GENERATION_EXPRESSION =
"delta.generationExpression";
private static final List<ParsedGeneratedExpr.GeneratedExprType>
GRANULARITIES =
Arrays.asList(
ParsedGeneratedExpr.GeneratedExprType.YEAR,
diff --git
a/xtable-core/src/main/java/org/apache/xtable/kernel/DeltaKernelSchemaExtractor.java
b/xtable-core/src/main/java/org/apache/xtable/kernel/DeltaKernelSchemaExtractor.java
index 464ec2bb..a51fa170 100644
---
a/xtable-core/src/main/java/org/apache/xtable/kernel/DeltaKernelSchemaExtractor.java
+++
b/xtable-core/src/main/java/org/apache/xtable/kernel/DeltaKernelSchemaExtractor.java
@@ -18,6 +18,14 @@
package org.apache.xtable.kernel;
+import static org.apache.xtable.delta.Constants.DELTA_COLUMN_MAPPING_ID;
+import static org.apache.xtable.delta.Constants.DELTA_COLUMN_MAPPING_NAME;
+import static
org.apache.xtable.delta.Constants.DELTA_COLUMN_MAPPING_NESTED_IDS;
+import static org.apache.xtable.delta.Constants.DELTA_GENERATION_EXPRESSION;
+import static
org.apache.xtable.delta.Constants.PARQUET_LIST_ELEMENT_FIELD_NAME;
+import static org.apache.xtable.delta.Constants.PARQUET_MAP_KEY_FIELD_NAME;
+import static org.apache.xtable.delta.Constants.PARQUET_MAP_VALUE_FIELD_NAME;
+
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
@@ -43,7 +51,6 @@ import io.delta.kernel.types.TimestampNTZType;
import io.delta.kernel.types.TimestampType;
import org.apache.xtable.collectors.CustomCollectors;
-import org.apache.xtable.delta.DeltaPartitionExtractor;
import org.apache.xtable.exception.NotSupportedException;
import org.apache.xtable.model.schema.InternalField;
import org.apache.xtable.model.schema.InternalSchema;
@@ -52,20 +59,29 @@ import org.apache.xtable.schema.SchemaUtils;
public class DeltaKernelSchemaExtractor {
- private static final String DELTA_COLUMN_MAPPING_ID =
"delta.columnMapping.id";
private static final DeltaKernelSchemaExtractor INSTANCE = new
DeltaKernelSchemaExtractor();
private static final Map<InternalSchema.MetadataKey, Object>
DEFAULT_TIMESTAMP_PRECISION_METADATA =
Collections.singletonMap(
InternalSchema.MetadataKey.TIMESTAMP_PRECISION,
InternalSchema.MetadataValue.MICROS);
- static final String DELTA_GENERATION_EXPRESSION =
"delta.generationExpression";
public static DeltaKernelSchemaExtractor getInstance() {
return INSTANCE;
}
public InternalSchema toInternalSchema(StructType structType) {
- return toInternalSchema(structType, null, false, null, null);
+ return toInternalSchema(structType, null, false, null, null, null, null);
+ }
+
+ private static String childIdPath(String nestedIdPath, String childName) {
+ return nestedIdPath == null ? null : nestedIdPath + "." + childName;
+ }
+
+ private static Integer nestedFieldId(FieldMetadata nestedIds, String path) {
+ if (nestedIds == null || path == null || !nestedIds.contains(path)) {
+ return null;
+ }
+ return nestedIds.getLong(path).intValue();
}
private InternalSchema toInternalSchema(
@@ -73,7 +89,9 @@ public class DeltaKernelSchemaExtractor {
String parentPath,
boolean nullable,
String comment,
- FieldMetadata originalMetadata) {
+ FieldMetadata originalMetadata,
+ FieldMetadata nestedIds,
+ String nestedIdPath) {
String trimmedTypeName = "";
InternalType type = null;
@@ -126,11 +144,7 @@ public class DeltaKernelSchemaExtractor {
fields =
structType.fields().stream()
- .filter(
- field ->
- !field
- .getMetadata()
-
.contains(DeltaPartitionExtractor.DELTA_GENERATION_EXPRESSION))
+ .filter(field ->
!field.getMetadata().contains(DELTA_GENERATION_EXPRESSION))
.map(
field -> {
Integer fieldId =
@@ -138,6 +152,10 @@ public class DeltaKernelSchemaExtractor {
?
Long.valueOf(field.getMetadata().getLong(DELTA_COLUMN_MAPPING_ID))
.intValue()
: null;
+ String storageName =
+ field.getMetadata().contains(DELTA_COLUMN_MAPPING_NAME)
+ ?
field.getMetadata().getString(DELTA_COLUMN_MAPPING_NAME)
+ : null;
String fieldComment =
field.getMetadata().contains("comment")
? field.getMetadata().getString("comment")
@@ -148,7 +166,11 @@ public class DeltaKernelSchemaExtractor {
SchemaUtils.getFullyQualifiedPath(parentPath,
field.getName()),
field.isNullable(),
fieldComment,
- field.getMetadata());
+ field.getMetadata(),
+
field.getMetadata().contains(DELTA_COLUMN_MAPPING_NESTED_IDS)
+ ?
field.getMetadata().getMetadata(DELTA_COLUMN_MAPPING_NESTED_IDS)
+ : null,
+ storageName != null ? storageName :
field.getName());
return InternalField.builder()
.name(field.getName())
.fieldId(fieldId)
@@ -171,6 +193,7 @@ public class DeltaKernelSchemaExtractor {
} else if (dataType instanceof ArrayType) {
ArrayType arrayType = (ArrayType) dataType;
+ String elementIdPath = childIdPath(nestedIdPath,
PARQUET_LIST_ELEMENT_FIELD_NAME);
InternalSchema elementSchema =
toInternalSchema(
arrayType.getElementType(),
@@ -178,11 +201,14 @@ public class DeltaKernelSchemaExtractor {
parentPath,
InternalField.Constants.ARRAY_ELEMENT_FIELD_NAME),
arrayType.containsNull(),
null,
- null);
+ null,
+ nestedIds,
+ elementIdPath);
InternalField elementField =
InternalField.builder()
.name(InternalField.Constants.ARRAY_ELEMENT_FIELD_NAME)
.parentPath(parentPath)
+ .fieldId(nestedFieldId(nestedIds, elementIdPath))
.schema(elementSchema)
.build();
type = InternalType.LIST;
@@ -190,6 +216,8 @@ public class DeltaKernelSchemaExtractor {
trimmedTypeName = "array";
} else if (dataType instanceof MapType) {
MapType mapType = (MapType) dataType;
+ String keyIdPath = childIdPath(nestedIdPath, PARQUET_MAP_KEY_FIELD_NAME);
+ String valueIdPath = childIdPath(nestedIdPath,
PARQUET_MAP_VALUE_FIELD_NAME);
InternalSchema keySchema =
toInternalSchema(
mapType.getKeyType(),
@@ -197,11 +225,14 @@ public class DeltaKernelSchemaExtractor {
parentPath, InternalField.Constants.MAP_KEY_FIELD_NAME),
false,
null,
- null);
+ null,
+ nestedIds,
+ keyIdPath);
InternalField keyField =
InternalField.builder()
.name(InternalField.Constants.MAP_KEY_FIELD_NAME)
.parentPath(parentPath)
+ .fieldId(nestedFieldId(nestedIds, keyIdPath))
.schema(keySchema)
.build();
InternalSchema valueSchema =
@@ -211,11 +242,14 @@ public class DeltaKernelSchemaExtractor {
parentPath, InternalField.Constants.MAP_VALUE_FIELD_NAME),
mapType.isValueContainsNull(),
null,
- null);
+ null,
+ nestedIds,
+ valueIdPath);
InternalField valueField =
InternalField.builder()
.name(InternalField.Constants.MAP_VALUE_FIELD_NAME)
.parentPath(parentPath)
+ .fieldId(nestedFieldId(nestedIds, valueIdPath))
.schema(valueSchema)
.build();
type = InternalType.MAP;
diff --git
a/xtable-core/src/test/java/org/apache/xtable/delta/TestDeltaPartitionExtractor.java
b/xtable-core/src/test/java/org/apache/xtable/delta/TestDeltaPartitionExtractor.java
index a405de08..0aaf3d60 100644
---
a/xtable-core/src/test/java/org/apache/xtable/delta/TestDeltaPartitionExtractor.java
+++
b/xtable-core/src/test/java/org/apache/xtable/delta/TestDeltaPartitionExtractor.java
@@ -18,7 +18,7 @@
package org.apache.xtable.delta;
-import static
org.apache.xtable.delta.DeltaPartitionExtractor.DELTA_GENERATION_EXPRESSION;
+import static org.apache.xtable.delta.Constants.DELTA_GENERATION_EXPRESSION;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
diff --git
a/xtable-core/src/test/java/org/apache/xtable/delta/TestDeltaSchemaExtractor.java
b/xtable-core/src/test/java/org/apache/xtable/delta/TestDeltaSchemaExtractor.java
index a8ca9410..a2e31920 100644
---
a/xtable-core/src/test/java/org/apache/xtable/delta/TestDeltaSchemaExtractor.java
+++
b/xtable-core/src/test/java/org/apache/xtable/delta/TestDeltaSchemaExtractor.java
@@ -18,9 +18,11 @@
package org.apache.xtable.delta;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import org.apache.spark.sql.types.DataTypes;
@@ -823,6 +825,180 @@ public class TestDeltaSchemaExtractor {
internalSchema,
DeltaSchemaExtractor.getInstance().toInternalSchema(structRepresentation));
}
+ @Test
+ public void testNestedFieldIdsInDeltaSchema() {
+ // Delta writes these ids for a collection's children when IcebergCompatV2
is enabled, keyed
+ // by the path the child takes in the parquet file.
+ Metadata mapMetadata =
+ Metadata.fromJson(
+ "{\"delta.columnMapping.id\": 1,
\"delta.columnMapping.physicalName\": \"col-map\","
+ + " \"delta.columnMapping.nested.ids\": {\"col-map.key\": 7,
\"col-map.value\": 8}}");
+ Metadata listMetadata =
+ Metadata.fromJson(
+ "{\"delta.columnMapping.id\": 2,
\"delta.columnMapping.physicalName\": \"col-list\","
+ + " \"delta.columnMapping.nested.ids\": {\"col-list.element\":
9}}");
+ Metadata plainMetadata =
+ Metadata.fromJson(
+ "{\"delta.columnMapping.id\": 3,
\"delta.columnMapping.physicalName\": \"col-plain\"}");
+ StructType structType =
+ new StructType()
+ .add(
+ "map_field",
+ DataTypes.createMapType(DataTypes.StringType,
DataTypes.IntegerType),
+ true,
+ mapMetadata)
+ .add("list_field",
DataTypes.createArrayType(DataTypes.IntegerType), true, listMetadata)
+ .add(
+ "plain_map",
+ DataTypes.createMapType(DataTypes.StringType,
DataTypes.IntegerType),
+ true,
+ plainMetadata);
+
+ InternalSchema internalSchema =
DeltaSchemaExtractor.getInstance().toInternalSchema(structType);
+
+ Assertions.assertEquals(7, fieldId(internalSchema, "map_field",
"_one_field_key"));
+ Assertions.assertEquals(8, fieldId(internalSchema, "map_field",
"_one_field_value"));
+ Assertions.assertEquals(9, fieldId(internalSchema, "list_field",
"_one_field_element"));
+ // A field without the metadata leaves its children unassigned, as before.
+ Assertions.assertNull(fieldId(internalSchema, "plain_map",
"_one_field_key"));
+ Assertions.assertNull(fieldId(internalSchema, "plain_map",
"_one_field_value"));
+ }
+
+ @Test
+ public void testNestedFieldIdsForComplexCollectionChildren() {
+ // Delta keys the nested ids by the path the child takes in the parquet
file, relative to the
+ // nearest parent struct field's physical name, so a collection nested in
a collection adds to
+ // the same field's metadata while a collection nested in a struct gets
metadata of its own.
+ Metadata structListMetadata =
+ Metadata.fromJson(
+ "{\"delta.columnMapping.id\": 1,
\"delta.columnMapping.physicalName\":"
+ + " \"col-struct-list\", \"delta.columnMapping.nested.ids\":"
+ + " {\"col-struct-list.element\": 20}}");
+ Metadata structMapMetadata =
+ Metadata.fromJson(
+ "{\"delta.columnMapping.id\": 2,
\"delta.columnMapping.physicalName\":"
+ + " \"col-struct-map\", \"delta.columnMapping.nested.ids\":"
+ + " {\"col-struct-map.key\": 30, \"col-struct-map.value\":
31}}");
+ Metadata nestedMapMetadata =
+ Metadata.fromJson(
+ "{\"delta.columnMapping.id\": 3,
\"delta.columnMapping.physicalName\":"
+ + " \"col-nested-map\", \"delta.columnMapping.nested.ids\":"
+ + " {\"col-nested-map.key\": 40, \"col-nested-map.value\": 41,"
+ + " \"col-nested-map.value.element\": 42}}");
+ Metadata memberListMetadata =
+ Metadata.fromJson(
+ "{\"delta.columnMapping.id\": 6,
\"delta.columnMapping.physicalName\":"
+ + " \"col-member-list\", \"delta.columnMapping.nested.ids\":"
+ + " {\"col-member-list.element\": 60}}");
+ StructType structType =
+ new StructType()
+ // list of structs: the element position carries the id, the
struct's members carry
+ // their own column mapping ids
+ .add(
+ "struct_list",
+ DataTypes.createArrayType(
+ new StructType()
+ .add(
+ "name",
+ DataTypes.StringType,
+ true,
+ Metadata.fromJson("{\"delta.columnMapping.id\":
21}"))
+ .add(
+ "quant",
+ DataTypes.IntegerType,
+ false,
+ Metadata.fromJson("{\"delta.columnMapping.id\":
22}"))),
+ true,
+ structListMetadata)
+ // map with a struct value: the key and value positions carry ids,
the value struct's
+ // members carry their own
+ .add(
+ "struct_map",
+ DataTypes.createMapType(
+ DataTypes.StringType,
+ new StructType()
+ .add(
+ "price",
+ DataTypes.DoubleType,
+ true,
+ Metadata.fromJson("{\"delta.columnMapping.id\":
32}"))),
+ true,
+ structMapMetadata)
+ // map of lists: with no struct field in between, all positions
stay keyed under the
+ // map field's own physical name
+ .add(
+ "nested_map",
+ DataTypes.createMapType(
+ DataTypes.StringType,
DataTypes.createArrayType(DataTypes.IntegerType)),
+ true,
+ nestedMapMetadata)
+ // map whose value struct holds a list: the list is a struct field
of its own, so its
+ // element id comes from the list field's own metadata, not from
the map's
+ .add(
+ "map_with_member_list",
+ DataTypes.createMapType(
+ DataTypes.StringType,
+ new StructType()
+ .add(
+ "tags",
+ DataTypes.createArrayType(DataTypes.StringType),
+ true,
+ memberListMetadata)),
+ true,
+ Metadata.fromJson(
+ "{\"delta.columnMapping.id\": 5,
\"delta.columnMapping.physicalName\":"
+ + " \"col-map-with-member-list\",
\"delta.columnMapping.nested.ids\":"
+ + " {\"col-map-with-member-list.key\": 50,"
+ + " \"col-map-with-member-list.value\": 51}}"));
+
+ InternalSchema internalSchema =
DeltaSchemaExtractor.getInstance().toInternalSchema(structType);
+
+ Assertions.assertEquals(20, fieldId(internalSchema, "struct_list",
"_one_field_element"));
+ Assertions.assertEquals(
+ 21, fieldId(internalSchema, "struct_list", "_one_field_element",
"name"));
+ Assertions.assertEquals(
+ 22, fieldId(internalSchema, "struct_list", "_one_field_element",
"quant"));
+ Assertions.assertEquals(30, fieldId(internalSchema, "struct_map",
"_one_field_key"));
+ Assertions.assertEquals(31, fieldId(internalSchema, "struct_map",
"_one_field_value"));
+ Assertions.assertEquals(32, fieldId(internalSchema, "struct_map",
"_one_field_value", "price"));
+ Assertions.assertEquals(40, fieldId(internalSchema, "nested_map",
"_one_field_key"));
+ Assertions.assertEquals(41, fieldId(internalSchema, "nested_map",
"_one_field_value"));
+ Assertions.assertEquals(
+ 42, fieldId(internalSchema, "nested_map", "_one_field_value",
"_one_field_element"));
+ Assertions.assertEquals(50, fieldId(internalSchema,
"map_with_member_list", "_one_field_key"));
+ Assertions.assertEquals(
+ 51, fieldId(internalSchema, "map_with_member_list",
"_one_field_value"));
+ Assertions.assertEquals(
+ 60,
+ fieldId(
+ internalSchema,
+ "map_with_member_list",
+ "_one_field_value",
+ "tags",
+ "_one_field_element"));
+ }
+
+ private static Integer fieldId(InternalSchema schema, String fieldName,
String childName) {
+ return fieldId(schema, fieldName, new String[] {childName});
+ }
+
+ private static Integer fieldId(InternalSchema schema, String fieldName,
String... childNames) {
+ InternalSchema current = schema;
+ InternalField found = null;
+ List<String> path = new ArrayList<>(childNames.length + 1);
+ path.add(fieldName);
+ path.addAll(Arrays.asList(childNames));
+ for (String name : path) {
+ found =
+ current.getFields().stream()
+ .filter(field -> name.equals(field.getName()))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("missing " + name));
+ current = found.getSchema();
+ }
+ return found.getFieldId();
+ }
+
@Test
public void testFieldIdsInDeltaSchema() {
StructType structRepresentation =
diff --git
a/xtable-core/src/test/java/org/apache/xtable/kernel/TestDeltaKernelPartitionExtractor.java
b/xtable-core/src/test/java/org/apache/xtable/kernel/TestDeltaKernelPartitionExtractor.java
index bfdbaf8f..e574f99b 100644
---
a/xtable-core/src/test/java/org/apache/xtable/kernel/TestDeltaKernelPartitionExtractor.java
+++
b/xtable-core/src/test/java/org/apache/xtable/kernel/TestDeltaKernelPartitionExtractor.java
@@ -18,7 +18,7 @@
package org.apache.xtable.kernel;
-import static
org.apache.xtable.kernel.DeltaKernelPartitionExtractor.DELTA_GENERATION_EXPRESSION;
+import static org.apache.xtable.delta.Constants.DELTA_GENERATION_EXPRESSION;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
diff --git
a/xtable-core/src/test/java/org/apache/xtable/kernel/TestDeltaKernelSchemaExtractor.java
b/xtable-core/src/test/java/org/apache/xtable/kernel/TestDeltaKernelSchemaExtractor.java
index 8c3dda57..7051ac7d 100644
---
a/xtable-core/src/test/java/org/apache/xtable/kernel/TestDeltaKernelSchemaExtractor.java
+++
b/xtable-core/src/test/java/org/apache/xtable/kernel/TestDeltaKernelSchemaExtractor.java
@@ -22,9 +22,11 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import org.junit.jupiter.api.Assertions;
@@ -319,6 +321,131 @@ public class TestDeltaKernelSchemaExtractor {
DeltaKernelSchemaExtractor.getInstance().toInternalSchema(structRepresentation));
}
+ @Test
+ public void testNestedFieldIdsInDeltaSchema() {
+ // Delta writes these ids for a collection's children when IcebergCompatV2
is enabled, keyed
+ // by the path the child takes in the parquet file, relative to the
nearest parent struct
+ // field's physical name.
+ FieldMetadata mapMetadata =
+ columnMappingMetadata(
+ 1,
+ "col-map",
+ nestedIds(
+ "col-map.key", 7L,
+ "col-map.value", 8L));
+ FieldMetadata listMetadata =
+ columnMappingMetadata(2, "col-list", nestedIds("col-list.element",
9L));
+ FieldMetadata nestedMapMetadata =
+ columnMappingMetadata(
+ 3,
+ "col-nested-map",
+ nestedIds(
+ "col-nested-map.key", 40L,
+ "col-nested-map.value", 41L,
+ "col-nested-map.value.element", 42L));
+ FieldMetadata memberListMetadata =
+ columnMappingMetadata(6, "col-member-list",
nestedIds("col-member-list.element", 60L));
+ io.delta.kernel.types.StructType structRepresentation =
+ new StructType()
+ .add(
+ "map_field",
+ new MapType(StringType.STRING, IntegerType.INTEGER, false),
+ true,
+ mapMetadata)
+ .add("list_field", new ArrayType(IntegerType.INTEGER, true), true,
listMetadata)
+ .add(
+ "plain_map",
+ new MapType(StringType.STRING, IntegerType.INTEGER, false),
+ true,
+ FieldMetadata.builder().putLong("delta.columnMapping.id",
4).build())
+ // map of lists: with no struct field in between, all positions
stay keyed under the
+ // map field's own physical name
+ .add(
+ "nested_map",
+ new MapType(StringType.STRING, new
ArrayType(IntegerType.INTEGER, true), false),
+ true,
+ nestedMapMetadata)
+ // map whose value struct holds a list: the list is a struct field
of its own, so its
+ // element id comes from the list field's own metadata, not from
the map's
+ .add(
+ "map_with_member_list",
+ new MapType(
+ StringType.STRING,
+ new StructType()
+ .add(
+ "tags",
+ new ArrayType(StringType.STRING, true),
+ true,
+ memberListMetadata),
+ false),
+ true,
+ columnMappingMetadata(
+ 5,
+ "col-map-with-member-list",
+ nestedIds(
+ "col-map-with-member-list.key", 50L,
+ "col-map-with-member-list.value", 51L)));
+
+ InternalSchema internalSchema =
+
DeltaKernelSchemaExtractor.getInstance().toInternalSchema(structRepresentation);
+
+ Assertions.assertEquals(7, fieldId(internalSchema, "map_field",
"_one_field_key"));
+ Assertions.assertEquals(8, fieldId(internalSchema, "map_field",
"_one_field_value"));
+ Assertions.assertEquals(9, fieldId(internalSchema, "list_field",
"_one_field_element"));
+ // A field without the metadata leaves its children unassigned, as before.
+ Assertions.assertNull(fieldId(internalSchema, "plain_map",
"_one_field_key"));
+ Assertions.assertNull(fieldId(internalSchema, "plain_map",
"_one_field_value"));
+ Assertions.assertEquals(40, fieldId(internalSchema, "nested_map",
"_one_field_key"));
+ Assertions.assertEquals(41, fieldId(internalSchema, "nested_map",
"_one_field_value"));
+ Assertions.assertEquals(
+ 42, fieldId(internalSchema, "nested_map", "_one_field_value",
"_one_field_element"));
+ Assertions.assertEquals(50, fieldId(internalSchema,
"map_with_member_list", "_one_field_key"));
+ Assertions.assertEquals(
+ 51, fieldId(internalSchema, "map_with_member_list",
"_one_field_value"));
+ Assertions.assertEquals(
+ 60,
+ fieldId(
+ internalSchema,
+ "map_with_member_list",
+ "_one_field_value",
+ "tags",
+ "_one_field_element"));
+ }
+
+ private static FieldMetadata columnMappingMetadata(
+ long id, String physicalName, FieldMetadata nestedIds) {
+ return FieldMetadata.builder()
+ .putLong("delta.columnMapping.id", id)
+ .putString("delta.columnMapping.physicalName", physicalName)
+ .putFieldMetadata("delta.columnMapping.nested.ids", nestedIds)
+ .build();
+ }
+
+ private static FieldMetadata nestedIds(Object... pathToId) {
+ FieldMetadata.Builder builder = FieldMetadata.builder();
+ for (int i = 0; i < pathToId.length; i += 2) {
+ builder.putLong((String) pathToId[i], (Long) pathToId[i + 1]);
+ }
+ return builder.build();
+ }
+
+ private static Integer fieldId(InternalSchema schema, String fieldName,
String... childNames) {
+ InternalSchema current = schema;
+ InternalField found = null;
+ List<String> path = new ArrayList<>(childNames.length + 1);
+ path.add(fieldName);
+ path.addAll(Arrays.asList(childNames));
+ for (String name : path) {
+ found =
+ current.getFields().stream()
+ .filter(field -> name.equals(field.getName()))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("missing " + name));
+ current = found.getSchema();
+ }
+ return found.getFieldId();
+ }
+
@Test
public void testTimestamps() {
Map<InternalSchema.MetadataKey, Object> metadata =
