Skip to content

Commit c5432c8

Browse files
ramitg254ramitg254
committed
HIVE-29341:Iceberg: [V3] NULL initial default for STRUCT column
Change-Id: Iefec29f3d2e320cd94e5944651d65e85471b5cb8
1 parent 1ea685e commit c5432c8

10 files changed

Lines changed: 349 additions & 64 deletions

File tree

‎iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/HiveSchemaUtil.java‎

Lines changed: 76 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -154,6 +154,23 @@ public static Type convert(TypeInfo typeInfo, String defaultValue) {
154154
return HiveSchemaConverter.convert(typeInfo, false, defaultValue);
155155
}
156156

157+
public static Type applyInitialDefaultsToStruct(Type type) {
158+
Types.StructType struct = type.asStructType();
159+
return Types.StructType.of(
160+
struct.fields().stream().map(HiveSchemaUtil::applyInitialDefaultsToField).collect(Collectors.toList()));
161+
}
162+
163+
private static Types.NestedField applyInitialDefaultsToField(Types.NestedField field) {
164+
Types.NestedField.Builder builder = Types.NestedField.from(field);
165+
if (field.type().isStructType()) {
166+
builder.ofType(applyInitialDefaultsToStruct(field.type()));
167+
}
168+
if (field.initialDefaultLiteral() == null && field.writeDefaultLiteral() != null) {
169+
builder.withInitialDefault(field.writeDefaultLiteral());
170+
}
171+
return builder.build();
172+
}
173+
157174
/**
158175
* Returns a SchemaDifference containing those fields which are present in only one of the collections, as well as
159176
* those fields which are present in both (in terms of the name) but their type or comment has changed.
@@ -429,6 +446,24 @@ public static void setDefaultValues(Record record, List<Types.NestedField> missi
429446
}
430447
}
431448

449+
/**
450+
* Backfills struct column that is null on read using nested {@code initialDefault} metadata.
451+
* This applies to rows written before {@code ADD COLUMNS} added the struct.
452+
* Spec allows struct defaults as {@code {}} (see https://iceberg.apache.org/spec/#default-values), but
453+
* {@code UpdateSchema} add column only supports primitives today;
454+
* if empty structs are allowed, this backfill can be removed.
455+
*/
456+
public static void backfillStructInitialDefaults(Record record, List<Types.NestedField> columns) {
457+
for (Types.NestedField field : columns) {
458+
if (field.type().isStructType() && record.getField(field.name()) == null) {
459+
Record nestedRecord = buildStructWithInitialDefaults(field.type().asStructType());
460+
if (nestedRecord != null) {
461+
record.setField(field.name(), nestedRecord);
462+
}
463+
}
464+
}
465+
}
466+
432467
/**
433468
* Recursively builds a struct populated with write defaults.
434469
* * @return A populated Record, or null if no nested fields have defaults.
@@ -458,6 +493,47 @@ private static Record buildStructWithDefaults(Types.StructType structType) {
458493
return hasAnyDefault ? nestedRecord : null;
459494
}
460495

496+
private static Record buildStructWithInitialDefaults(Types.StructType structType) {
497+
Record nestedRecord = GenericRecord.create(structType);
498+
boolean hasAnyDefault = false;
499+
500+
for (Types.NestedField field : structType.fields()) {
501+
if (field.initialDefault() != null) {
502+
Object defaultValue = convertToWriteType(field.initialDefault(), field.type());
503+
nestedRecord.setField(field.name(), defaultValue);
504+
hasAnyDefault = true;
505+
} else if (field.type().isStructType()) {
506+
Record deeperRecord = buildStructWithInitialDefaults(field.type().asStructType());
507+
if (deeperRecord != null) {
508+
nestedRecord.setField(field.name(), deeperRecord);
509+
hasAnyDefault = true;
510+
}
511+
}
512+
}
513+
514+
return hasAnyDefault ? nestedRecord : null;
515+
}
516+
517+
/**
518+
* Builds a map of nested field names to their initial-default values for a struct column.
519+
*
520+
* @return field-name to initial-default map, or empty if no nested field has an initial-default
521+
*/
522+
public static Map<String, Object> getStructInitialDefaults(Types.StructType structType) {
523+
Map<String, Object> result = Maps.newHashMap();
524+
for (Types.NestedField field : structType.fields()) {
525+
if (field.initialDefault() != null) {
526+
result.put(field.name(), field.initialDefault());
527+
} else if (field.type().isStructType()) {
528+
Map<String, Object> nested = getStructInitialDefaults(field.type().asStructType());
529+
if (!nested.isEmpty()) {
530+
result.put(field.name(), nested);
531+
}
532+
}
533+
}
534+
return result;
535+
}
536+
461537
/**
462538
* Sets a value into a {@link Record} using a struct-only field path (top-level column or nested
463539
* through structs). Intermediate struct records are created as needed.
@@ -496,21 +572,6 @@ private static Record getOrCreateStructRecord(
496572
return record;
497573
}
498574

499-
// Special method for nested structs that always applies defaults to null fields
500-
private static void setDefaultValuesForNestedStruct(Record record, List<Types.NestedField> fields) {
501-
for (Types.NestedField field : fields) {
502-
Object fieldValue = record.getField(field.name());
503-
504-
if (field.writeDefault() != null) {
505-
Object defaultValue = convertToWriteType(field.writeDefault(), field.type());
506-
record.setField(field.name(), defaultValue);
507-
} else if (field.type().isStructType()) {
508-
// Recursively process nested structs
509-
setDefaultValuesForNestedStruct((Record) fieldValue, field.type().asStructType().fields());
510-
}
511-
}
512-
}
513-
514575
public static Object convertToWriteType(Object value, Type type) {
515576
if (value == null) {
516577
return null;

‎iceberg/iceberg-catalog/src/test/java/org/apache/iceberg/hive/TestHiveSchemaUtil.java‎

Lines changed: 103 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,12 +22,16 @@
2222
import java.util.Arrays;
2323
import java.util.Collections;
2424
import java.util.List;
25+
import java.util.Map;
2526
import java.util.stream.Collectors;
2627
import org.apache.hadoop.hive.metastore.api.FieldSchema;
2728
import org.apache.hadoop.hive.serde.serdeConstants;
2829
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
2930
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;
3031
import org.apache.iceberg.Schema;
32+
import org.apache.iceberg.data.GenericRecord;
33+
import org.apache.iceberg.data.Record;
34+
import org.apache.iceberg.expressions.Expressions;
3135
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
3236
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
3337
import org.apache.iceberg.types.Type;
@@ -37,6 +41,7 @@
3741
import static org.apache.iceberg.types.Types.NestedField.optional;
3842
import static org.assertj.core.api.Assertions.assertThat;
3943
import static org.assertj.core.api.Assertions.assertThatThrownBy;
44+
import static org.assertj.core.api.InstanceOfAssertFactories.MAP;
4045

4146
public class TestHiveSchemaUtil {
4247
private static final Schema SIMPLE_ICEBERG_SCHEMA = new Schema(
@@ -240,4 +245,102 @@ private void assertEquals(Type expected, Type actual) {
240245
}
241246
}
242247
}
248+
249+
@Test
250+
public void testBackfillStructInitialDefaults() {
251+
Schema schema = new Schema(
252+
optional(1, "id", Types.IntegerType.get()),
253+
optional(2, "point", Types.StructType.of(
254+
Types.NestedField.builder()
255+
.withId(3)
256+
.withName("x")
257+
.asOptional()
258+
.ofType(Types.IntegerType.get())
259+
.withInitialDefault(Expressions.lit(100))
260+
.build(),
261+
Types.NestedField.builder()
262+
.withId(4)
263+
.withName("y")
264+
.asOptional()
265+
.ofType(Types.IntegerType.get())
266+
.withInitialDefault(Expressions.lit(99))
267+
.build()
268+
))
269+
);
270+
271+
Record record = GenericRecord.create(schema);
272+
record.setField("id", 1);
273+
274+
HiveSchemaUtil.backfillStructInitialDefaults(record, schema.columns());
275+
276+
Record point = (Record) record.getField("point");
277+
assertThat(point).isNotNull();
278+
assertThat(point.getField("x")).isEqualTo(100);
279+
assertThat(point.getField("y")).isEqualTo(99);
280+
}
281+
282+
@Test
283+
public void testApplyInitialDefaults() {
284+
Type structType = Types.StructType.of(
285+
Types.NestedField.builder()
286+
.withId(1)
287+
.withName("x")
288+
.asOptional()
289+
.ofType(Types.IntegerType.get())
290+
.withWriteDefault(Expressions.lit(100))
291+
.build(),
292+
Types.NestedField.builder()
293+
.withId(2)
294+
.withName("y")
295+
.asOptional()
296+
.ofType(Types.IntegerType.get())
297+
.withWriteDefault(Expressions.lit(99))
298+
.build());
299+
300+
Type withInitialDefaults = HiveSchemaUtil.applyInitialDefaultsToStruct(structType);
301+
Types.NestedField xField = withInitialDefaults.asStructType().field("x");
302+
Types.NestedField yField = withInitialDefaults.asStructType().field("y");
303+
assertThat(xField.initialDefault()).isEqualTo(100);
304+
assertThat(yField.initialDefault()).isEqualTo(99);
305+
}
306+
307+
@Test
308+
public void testGetStructInitialDefaults() {
309+
Types.StructType addressStruct = Types.StructType.of(
310+
Types.NestedField.builder()
311+
.withId(4)
312+
.withName("street")
313+
.asOptional()
314+
.ofType(Types.StringType.get())
315+
.withInitialDefault(Expressions.lit("Main St"))
316+
.build(),
317+
Types.NestedField.builder()
318+
.withId(5)
319+
.withName("city")
320+
.asOptional()
321+
.ofType(Types.StringType.get())
322+
.withInitialDefault(Expressions.lit("New York"))
323+
.build());
324+
Types.StructType personStruct = Types.StructType.of(
325+
Types.NestedField.builder()
326+
.withId(3)
327+
.withName("name")
328+
.asOptional()
329+
.ofType(Types.StringType.get())
330+
.withInitialDefault(Expressions.lit("John"))
331+
.build(),
332+
Types.NestedField.builder()
333+
.withId(6)
334+
.withName("address")
335+
.asOptional()
336+
.ofType(addressStruct)
337+
.build());
338+
339+
Map<String, Object> personDefaults = HiveSchemaUtil.getStructInitialDefaults(personStruct);
340+
assertThat(personDefaults)
341+
.containsEntry("name", "John")
342+
.extractingByKey("address", MAP)
343+
.containsEntry("street", "Main St")
344+
.containsEntry("city", "New York");
345+
}
243346
}

‎iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergMetaHook.java‎

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -744,12 +744,14 @@ private void handleAddColumns(org.apache.hadoop.hive.metastore.api.Table hmsTabl
744744
boolean isORc = isOrcFileFormat(hmsTable);
745745
for (FieldSchema addedCol : addedCols) {
746746
String defaultValue = defaultValues.get(addedCol.getName());
747-
Type type = HiveSchemaUtil.convert(TypeInfoUtils.getTypeInfoFromTypeString(addedCol.getType()), defaultValue);
748-
Literal<Object> defaultVal = Optional.ofNullable(defaultValue).filter(v -> !type.isStructType())
749-
.map(v -> Expressions.lit(HiveSchemaUtil.getDefaultValue(v, type))).orElse(null);
750-
747+
Type baseType = HiveSchemaUtil.convert(TypeInfoUtils.getTypeInfoFromTypeString(addedCol.getType()), defaultValue);
748+
final Type resolvedType = (!isORc && baseType.isStructType()) ?
749+
HiveSchemaUtil.applyInitialDefaultsToStruct(baseType) :
750+
baseType;
751+
Literal<Object> defaultVal = Optional.ofNullable(defaultValue).filter(v -> !resolvedType.isStructType())
752+
.map(v -> Expressions.lit(HiveSchemaUtil.getDefaultValue(v, resolvedType))).orElse(null);
751753
// ORC doesn't have support for initialDefault from iceberg layer, we only need to set default for writeDefault.
752-
updateSchema.addColumn(addedCol.getName(), type, addedCol.getComment(), isORc ? null : defaultVal);
754+
updateSchema.addColumn(addedCol.getName(), resolvedType, addedCol.getComment(), isORc ? null : defaultVal);
753755
if (isORc && defaultVal != null) {
754756
updateSchema.updateColumnDefault(addedCol.getName(), defaultVal);
755757
}

‎iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vector/HiveVectorizedReader.java‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,7 @@
5252
import org.apache.iceberg.Schema;
5353
import org.apache.iceberg.Table;
5454
import org.apache.iceberg.expressions.Expression;
55+
import org.apache.iceberg.hive.HiveSchemaUtil;
5556
import org.apache.iceberg.io.CloseableIterable;
5657
import org.apache.iceberg.io.CloseableIterator;
5758
import org.apache.iceberg.mr.InputFormatConfig;
@@ -187,6 +188,12 @@ static Map<String, Object> getInitialColumnDefaults(List<Types.NestedField> colu
187188
for (Types.NestedField column : columns) {
188189
if (column.initialDefault() != null) {
189190
columnDefaults.put(column.name(), column.initialDefault());
191+
} else if (column.type().isStructType()) {
192+
Map<String, Object> structDefaults =
193+
HiveSchemaUtil.getStructInitialDefaults(column.type().asStructType());
194+
if (!structDefaults.isEmpty()) {
195+
columnDefaults.put(column.name(), structDefaults);
196+
}
190197
}
191198
}
192199
return columnDefaults;

‎iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergRecordReader.java‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,7 @@
5353
import org.apache.iceberg.data.parquet.GenericParquetReaders;
5454
import org.apache.iceberg.encryption.EncryptedFiles;
5555
import org.apache.iceberg.expressions.Expression;
56+
import org.apache.iceberg.hive.HiveSchemaUtil;
5657
import org.apache.iceberg.io.CloseableIterable;
5758
import org.apache.iceberg.io.CloseableIterator;
5859
import org.apache.iceberg.io.InputFile;
@@ -173,7 +174,16 @@ private CloseableIterable openGeneric(FileScanTask task, Schema readSchema) {
173174
default -> throw new UnsupportedOperationException(
174175
String.format("Cannot read %s file: %s", file.format().name(), file.location()));
175176
};
176-
return applyResidualFiltering(iterable, residual, readSchema);
177+
return applyResidualFiltering(withStructInitialDefaultBackfill(iterable, readSchema), residual, readSchema);
178+
}
179+
180+
private CloseableIterable<T> withStructInitialDefaultBackfill(CloseableIterable<T> iterable, Schema readSchema) {
181+
return CloseableIterable.transform(iterable, record -> {
182+
if (record instanceof Record) {
183+
HiveSchemaUtil.backfillStructInitialDefaults((Record) record, readSchema.columns());
184+
}
185+
return record;
186+
});
177187
}
178188

179189
private CloseableIterable<T> newAvroIterable(

‎iceberg/iceberg-handler/src/test/queries/positive/iceberg_initial_default.q‎

Lines changed: 24 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,14 @@ ALTER TABLE ice_parq ADD COLUMNS (point STRUCT<x:INT, y:INT> DEFAULT '{"x":100,"
1313
created_date DATE DEFAULT '2024-01-01',
1414
created_ts TIMESTAMP DEFAULT '2024-01-01T10:00:00',
1515
score DECIMAL(5,2) DEFAULT 100.00,
16-
category STRING DEFAULT 'general');
16+
category STRING DEFAULT 'general',
17+
person STRUCT<
18+
name: STRING,
19+
address: STRUCT<
20+
street: STRING,
21+
city: STRING
22+
>
23+
> DEFAULT '{"name":"John","address":{"street":"Main St","city":"New York"}}');
1724

1825
INSERT INTO ice_parq (id) VALUES (2);
1926

@@ -45,7 +52,14 @@ ALTER TABLE ice_avro ADD COLUMNS (point STRUCT<x:INT, y:INT> DEFAULT '{"x":100,"
4552
created_date DATE DEFAULT '2024-01-01',
4653
created_ts TIMESTAMP DEFAULT '2024-01-01T10:00:00',
4754
score DECIMAL(5,2) DEFAULT 100.00,
48-
category STRING DEFAULT 'general');
55+
category STRING DEFAULT 'general',
56+
person STRUCT<
57+
name: STRING,
58+
address: STRUCT<
59+
street: STRING,
60+
city: STRING
61+
>
62+
> DEFAULT '{"name":"John","address":{"street":"Main St","city":"New York"}}');
4963

5064
INSERT INTO ice_avro (id) VALUES (2);
5165

@@ -77,7 +91,14 @@ ALTER TABLE ice_orc ADD COLUMNS (point STRUCT<x:INT, y:INT> DEFAULT '{"x":100,"y
7791
created_date DATE DEFAULT '2024-01-01',
7892
created_ts TIMESTAMP DEFAULT '2024-01-01T10:00:00',
7993
score DECIMAL(5,2) DEFAULT 100.00,
80-
category STRING DEFAULT 'general');
94+
category STRING DEFAULT 'general',
95+
person STRUCT<
96+
name: STRING,
97+
address: STRUCT<
98+
street: STRING,
99+
city: STRING
100+
>
101+
> DEFAULT '{"name":"John","address":{"street":"Main St","city":"New York"}}');
81102

82103
INSERT INTO ice_orc (id) VALUES (2);
83104

‎iceberg/iceberg-handler/src/test/results/positive/iceberg_alter_default_column.q.out‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -60,7 +60,7 @@ POSTHOOK: query: SELECT * FROM ice_t ORDER BY id
6060
POSTHOOK: type: QUERY
6161
POSTHOOK: Input: default@ice_t
6262
POSTHOOK: Output: hdfs://### HDFS PATH ###
63-
1 NULL unknown 25 50000.0 true 2024-01-01 2024-01-01 10:00:00 100.00 general
63+
1 {"x":100,"y":99} unknown 25 50000.0 true 2024-01-01 2024-01-01 10:00:00 100.00 general
6464
2 {"x":100,"y":99} unknown 25 50000.0 true 2024-01-01 2024-01-01 10:00:00 100.00 general
6565
PREHOOK: query: ALTER TABLE ice_t REPLACE COLUMNS (id INT,
6666
point STRUCT<x:INT, y:INT> DEFAULT '{"x":100,"y":99}',
@@ -94,7 +94,7 @@ POSTHOOK: query: SELECT * FROM ice_t ORDER BY id
9494
POSTHOOK: type: QUERY
9595
POSTHOOK: Input: default@ice_t
9696
POSTHOOK: Output: hdfs://### HDFS PATH ###
97-
1 NULL unknown 25 50000.0 true 2024-01-01 2024-01-01 10:00:00 general
97+
1 {"x":100,"y":99} unknown 25 50000.0 true 2024-01-01 2024-01-01 10:00:00 general
9898
2 {"x":100,"y":99} unknown 25 50000.0 true 2024-01-01 2024-01-01 10:00:00 general
9999
PREHOOK: query: ALTER TABLE ice_t CHANGE COLUMN point point STRUCT<x:INT, y:INT> DEFAULT '{"x":100,"y":88}'
100100
PREHOOK: type: ALTERTABLE_RENAMECOL
@@ -128,7 +128,7 @@ POSTHOOK: query: SELECT * FROM ice_t ORDER BY id
128128
POSTHOOK: type: QUERY
129129
POSTHOOK: Input: default@ice_t
130130
POSTHOOK: Output: hdfs://### HDFS PATH ###
131-
1 NULL unknown 25 50000.0 true 2024-01-01 2024-01-01 10:00:00 general
131+
1 {"x":100,"y":99} unknown 25 50000.0 true 2024-01-01 2024-01-01 10:00:00 general
132132
2 {"x":100,"y":99} unknown 25 50000.0 true 2024-01-01 2024-01-01 10:00:00 general
133133
3 {"x":100,"y":88} unknown 21 50000.0 true 2024-01-01 2024-01-01 10:00:00 general
134134
PREHOOK: query: ALTER TABLE ice_t CHANGE COLUMN point point_new STRUCT<x:INT, y:INT> DEFAULT '{"x":55,"y":88}'
@@ -155,7 +155,7 @@ POSTHOOK: query: SELECT * FROM ice_t ORDER BY id
155155
POSTHOOK: type: QUERY
156156
POSTHOOK: Input: default@ice_t
157157
POSTHOOK: Output: hdfs://### HDFS PATH ###
158-
1 NULL unknown 25 50000.0 true 2024-01-01 2024-01-01 10:00:00 general
158+
1 {"x":100,"y":99} unknown 25 50000.0 true 2024-01-01 2024-01-01 10:00:00 general
159159
2 {"x":100,"y":99} unknown 25 50000.0 true 2024-01-01 2024-01-01 10:00:00 general
160160
3 {"x":100,"y":88} unknown 21 50000.0 true 2024-01-01 2024-01-01 10:00:00 general
161161
4 {"x":55,"y":88} unknown 21 50000.0 true 2024-01-01 2024-01-01 10:00:00 general

0 commit comments

Comments
 (0)