Skip to content

Commit 77e7dbb

Browse files
GuosmilesmileGuosmilesmile
authored
ORC: Add _row_id and _last_updated_sequence_number raeder in Orc to support lineage (#15776)
1 parent 57b1211 commit 77e7dbb

12 files changed

Lines changed: 357 additions & 46 deletions

File tree

‎data/src/test/java/org/apache/iceberg/data/orc/TestGenericData.java‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,11 @@ protected boolean supportsUnknown() {
6868
return true;
6969
}
7070

71+
@Override
72+
protected boolean supportsRowLineage() {
73+
return true;
74+
}
75+
7176
/** Orc writers don't have notion of non-null / required fields. */
7277
@Override
7378
protected boolean allowsWritingNullValuesForRequiredFields() {
@@ -250,13 +255,15 @@ private void writeAndValidateRecords(Schema schema, List<Record> expected) throw
250255
try (CloseableIterable<Record> reader =
251256
ORC.read(Files.localInput(testFile))
252257
.project(schema)
253-
.createReaderFunc(fileSchema -> GenericOrcReader.buildReader(schema, fileSchema))
258+
.createReaderFunc(
259+
fileSchema -> GenericOrcReader.buildReader(schema, fileSchema, ID_TO_CONSTANT))
254260
.build()) {
255261
rows = Lists.newArrayList(reader);
256262
}
257263

258264
for (int i = 0; i < expected.size(); i += 1) {
259-
DataTestHelpers.assertEquals(schema.asStruct(), expected.get(i), rows.get(i));
265+
DataTestHelpers.assertEquals(
266+
schema.asStruct(), expected.get(i), rows.get(i), ID_TO_CONSTANT, i);
260267
}
261268
}
262269
}

‎flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/data/FlinkOrcReader.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -70,7 +70,7 @@ public OrcValueReader<RowData> record(
7070
TypeDescription record,
7171
List<String> names,
7272
List<OrcValueReader<?>> fields) {
73-
return FlinkOrcReaders.struct(fields, iStruct, idToConstant);
73+
return FlinkOrcReaders.struct(record, fields, iStruct, idToConstant);
7474
}
7575

7676
@Override

‎flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/data/FlinkOrcReaders.java‎

Lines changed: 11 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@
3939
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
4040
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
4141
import org.apache.iceberg.types.Types;
42+
import org.apache.orc.TypeDescription;
4243
import org.apache.orc.storage.ql.exec.vector.BytesColumnVector;
4344
import org.apache.orc.storage.ql.exec.vector.ColumnVector;
4445
import org.apache.orc.storage.ql.exec.vector.DecimalColumnVector;
@@ -91,8 +92,11 @@ public static <K, V> OrcValueReader<MapData> map(
9192
}
9293

9394
public static OrcValueReader<RowData> struct(
94-
List<OrcValueReader<?>> readers, Types.StructType struct, Map<Integer, ?> idToConstant) {
95-
return new StructReader(readers, struct, idToConstant);
95+
TypeDescription record,
96+
List<OrcValueReader<?>> readers,
97+
Types.StructType struct,
98+
Map<Integer, ?> idToConstant) {
99+
return new StructReader(record, readers, struct, idToConstant);
96100
}
97101

98102
private static class StringReader implements OrcValueReader<StringData> {
@@ -265,8 +269,11 @@ private static class StructReader extends OrcValueReaders.StructReader<RowData>
265269
private final int numFields;
266270

267271
StructReader(
268-
List<OrcValueReader<?>> readers, Types.StructType struct, Map<Integer, ?> idToConstant) {
269-
super(readers, struct, idToConstant);
272+
TypeDescription record,
273+
List<OrcValueReader<?>> readers,
274+
Types.StructType struct,
275+
Map<Integer, ?> idToConstant) {
276+
super(record, readers, struct, idToConstant);
270277
this.numFields = struct.fields().size();
271278
}
272279

‎flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestRewriteDataFiles.java‎

Lines changed: 31 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@
3333
import java.util.List;
3434
import java.util.stream.StreamSupport;
3535
import org.apache.flink.streaming.api.graph.StreamGraphGenerator;
36+
import org.apache.iceberg.FileFormat;
3637
import org.apache.iceberg.ManifestFiles;
3738
import org.apache.iceberg.MetadataColumns;
3839
import org.apache.iceberg.Schema;
@@ -44,8 +45,14 @@
4445
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
4546
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
4647
import org.junit.jupiter.api.Test;
48+
import org.junit.jupiter.params.ParameterizedTest;
49+
import org.junit.jupiter.params.provider.FieldSource;
4750

4851
class TestRewriteDataFiles extends MaintenanceTaskTestBase {
52+
53+
private static final FileFormat[] FILE_FORMATS =
54+
new FileFormat[] {FileFormat.AVRO, FileFormat.PARQUET, FileFormat.ORC};
55+
4956
@Test
5057
void testRewriteUnpartitioned() throws Exception {
5158
Table table = createTable();
@@ -83,13 +90,14 @@ void testRewriteUnpartitioned() throws Exception {
8390
createRecord(4, "d")));
8491
}
8592

86-
@Test
87-
void testRewriteUnpartitionedPreserveLineage() throws Exception {
88-
Table table = createTable(3);
89-
insert(table, 1, "a");
90-
insert(table, 2, "b");
91-
insert(table, 3, "c");
92-
insert(table, 4, "d");
93+
@ParameterizedTest
94+
@FieldSource("FILE_FORMATS")
95+
void testRewriteUnpartitionedPreserveLineage(FileFormat fileFormat) throws Exception {
96+
Table table = createTable(3, fileFormat);
97+
insert(table, 1, "a", fileFormat);
98+
insert(table, 2, "b", fileFormat);
99+
insert(table, 3, "c", fileFormat);
100+
insert(table, 4, "d", fileFormat);
93101

94102
assertFileNum(table, 4, 0);
95103

@@ -123,15 +131,17 @@ void testRewriteUnpartitionedPreserveLineage() throws Exception {
123131
schema);
124132
}
125133

126-
@Test
127-
void testRewriteTheSameFilePreserveLineage() throws Exception {
128-
Table table = createTable(3);
129-
insert(table, 1, "a");
130-
insert(table, 2, "b");
134+
@ParameterizedTest
135+
@FieldSource("FILE_FORMATS")
136+
void testRewriteTheSameFilePreserveLineage(FileFormat fileFormat) throws Exception {
137+
Table table = createTable(3, fileFormat);
138+
insert(table, 1, "a", fileFormat);
139+
insert(table, 2, "b", fileFormat);
131140
// Create a file with two lines of data to verify that the rowid is read correctly.
132141
insert(
133142
table,
134-
ImmutableList.of(SimpleDataUtil.createRecord(3, "c"), SimpleDataUtil.createRecord(4, "d")));
143+
ImmutableList.of(SimpleDataUtil.createRecord(3, "c"), SimpleDataUtil.createRecord(4, "d")),
144+
fileFormat);
135145

136146
assertFileNum(table, 3, 0);
137147

@@ -167,13 +177,14 @@ void testRewriteTheSameFilePreserveLineage() throws Exception {
167177
schema);
168178
}
169179

170-
@Test
171-
void testRewritePartitionedPreserveLineage() throws Exception {
172-
Table table = createPartitionedTable(3);
173-
insertPartitioned(table, 1, "p1");
174-
insertPartitioned(table, 2, "p1");
175-
insertPartitioned(table, 3, "p2");
176-
insertPartitioned(table, 4, "p2");
180+
@ParameterizedTest
181+
@FieldSource("FILE_FORMATS")
182+
void testRewritePartitionedPreserveLineage(FileFormat fileFormat) throws Exception {
183+
Table table = createPartitionedTable(3, fileFormat);
184+
insertPartitioned(table, 1, "p1", fileFormat);
185+
insertPartitioned(table, 2, "p1", fileFormat);
186+
insertPartitioned(table, 3, "p2", fileFormat);
187+
insertPartitioned(table, 4, "p2", fileFormat);
177188

178189
assertFileNum(table, 4, 0);
179190

‎flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/OperatorTestBase.java‎

Lines changed: 29 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -133,10 +133,14 @@ void after() throws IOException {
133133
}
134134

135135
protected static Table createTable() {
136-
return createTable(2);
136+
return createTable(2, FileFormat.PARQUET);
137137
}
138138

139139
protected static Table createTable(int formatVersion) {
140+
return createPartitionedTable(formatVersion, FileFormat.PARQUET);
141+
}
142+
143+
protected static Table createTable(int formatVersion, FileFormat fileFormat) {
140144
return CATALOG_EXTENSION
141145
.catalog()
142146
.createTable(
@@ -145,6 +149,8 @@ protected static Table createTable(int formatVersion) {
145149
PartitionSpec.unpartitioned(),
146150
null,
147151
ImmutableMap.of(
152+
"write.format.default",
153+
fileFormat.name(),
148154
TableProperties.FORMAT_VERSION,
149155
String.valueOf(formatVersion),
150156
"flink.max-continuous-empty-commits",
@@ -182,7 +188,7 @@ protected static Table createTableWithDelete(int formatVersion) {
182188
"format-version", String.valueOf(formatVersion), "write.upsert.enabled", "true"));
183189
}
184190

185-
protected static Table createPartitionedTable(int formatVersion) {
191+
protected static Table createPartitionedTable(int formatVersion, FileFormat fileFormat) {
186192
return CATALOG_EXTENSION
187193
.catalog()
188194
.createTable(
@@ -191,24 +197,36 @@ protected static Table createPartitionedTable(int formatVersion) {
191197
PartitionSpec.builderFor(SimpleDataUtil.SCHEMA).identity("data").build(),
192198
null,
193199
ImmutableMap.of(
200+
"write.format.default",
201+
fileFormat.name(),
194202
"format-version",
195203
String.valueOf(formatVersion),
196204
"flink.max-continuous-empty-commits",
197205
"100000"));
198206
}
199207

200208
protected static Table createPartitionedTable() {
201-
return createPartitionedTable(2);
209+
return createPartitionedTable(2, FileFormat.PARQUET);
202210
}
203211

204212
protected void insert(Table table, Integer id, String data) throws IOException {
205-
new GenericAppenderHelper(table, FileFormat.PARQUET, warehouseDir)
213+
insert(table, id, data, FileFormat.PARQUET);
214+
}
215+
216+
protected void insert(Table table, Integer id, String data, FileFormat fileFormat)
217+
throws IOException {
218+
new GenericAppenderHelper(table, fileFormat, warehouseDir)
206219
.appendToTable(Lists.newArrayList(SimpleDataUtil.createRecord(id, data)));
207220
table.refresh();
208221
}
209222

210223
protected void insert(Table table, List<Record> records) throws IOException {
211-
new GenericAppenderHelper(table, FileFormat.PARQUET, warehouseDir).appendToTable(records);
224+
insert(table, records, FileFormat.PARQUET);
225+
}
226+
227+
protected void insert(Table table, List<Record> records, FileFormat fileFormat)
228+
throws IOException {
229+
new GenericAppenderHelper(table, fileFormat, warehouseDir).appendToTable(records);
212230
table.refresh();
213231
}
214232

@@ -309,7 +327,12 @@ protected void update(Table table, Integer id, String oldData, String tempData,
309327
}
310328

311329
protected void insertPartitioned(Table table, Integer id, String data) throws IOException {
312-
new GenericAppenderHelper(table, FileFormat.PARQUET, warehouseDir)
330+
insertPartitioned(table, id, data, FileFormat.PARQUET);
331+
}
332+
333+
protected void insertPartitioned(Table table, Integer id, String data, FileFormat fileFormat)
334+
throws IOException {
335+
new GenericAppenderHelper(table, fileFormat, warehouseDir)
313336
.appendToTable(
314337
TestHelpers.Row.of(data), Lists.newArrayList(SimpleDataUtil.createRecord(id, data)));
315338
table.refresh();

‎orc/src/main/java/org/apache/iceberg/data/orc/GenericOrcReader.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -76,7 +76,7 @@ public OrcValueReader<?> record(
7676
TypeDescription record,
7777
List<String> names,
7878
List<OrcValueReader<?>> fields) {
79-
return GenericOrcReaders.struct(fields, expected, idToConstant);
79+
return GenericOrcReaders.struct(record, fields, expected, idToConstant);
8080
}
8181

8282
@Override

‎orc/src/main/java/org/apache/iceberg/data/orc/GenericOrcReaders.java‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@
4343
import org.apache.iceberg.variants.Variant;
4444
import org.apache.iceberg.variants.VariantMetadata;
4545
import org.apache.iceberg.variants.VariantValue;
46+
import org.apache.orc.TypeDescription;
4647
import org.apache.orc.storage.ql.exec.vector.BytesColumnVector;
4748
import org.apache.orc.storage.ql.exec.vector.ColumnVector;
4849
import org.apache.orc.storage.ql.exec.vector.DecimalColumnVector;
@@ -56,11 +57,25 @@ public class GenericOrcReaders {
5657

5758
private GenericOrcReaders() {}
5859

60+
/**
61+
* @deprecated Use {@link #struct(TypeDescription, List, Types.StructType, Map)} instead. This
62+
* method uses position-based binding which may cause field misalignment in MOR and lineage
63+
* scenarios.
64+
*/
65+
@Deprecated
5966
public static OrcValueReader<Record> struct(
6067
List<OrcValueReader<?>> readers, Types.StructType struct, Map<Integer, ?> idToConstant) {
6168
return new StructReader(readers, struct, idToConstant);
6269
}
6370

71+
public static OrcValueReader<Record> struct(
72+
TypeDescription orcType,
73+
List<OrcValueReader<?>> readers,
74+
Types.StructType struct,
75+
Map<Integer, ?> idToConstant) {
76+
return new StructReader(orcType, readers, struct, idToConstant);
77+
}
78+
6479
public static OrcValueReader<List<?>> array(OrcValueReader<?> elementReader) {
6580
return new ListReader(elementReader);
6681
}
@@ -231,6 +246,12 @@ public Variant nonNullRead(ColumnVector vector, int row) {
231246
private static class StructReader extends OrcValueReaders.StructReader<Record> {
232247
private final GenericRecord template;
233248

249+
/**
250+
* @deprecated Use {@link #StructReader(TypeDescription, List, Types.StructType, Map)} instead.
251+
* This constructor uses position-based binding which may cause field misalignment in MOR
252+
* and lineage scenarios.
253+
*/
254+
@Deprecated
234255
protected StructReader(
235256
List<OrcValueReader<?>> readers,
236257
Types.StructType structType,
@@ -239,6 +260,15 @@ protected StructReader(
239260
this.template = GenericRecord.create(structType);
240261
}
241262

263+
protected StructReader(
264+
TypeDescription orcType,
265+
List<OrcValueReader<?>> readers,
266+
Types.StructType structType,
267+
Map<Integer, ?> idToConstant) {
268+
super(orcType, readers, structType, idToConstant);
269+
this.template = GenericRecord.create(structType);
270+
}
271+
242272
@Override
243273
protected Record create() {
244274
// GenericRecord.copy() is more performant then GenericRecord.create(StructType) since

‎orc/src/main/java/org/apache/iceberg/orc/ORC.java‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -787,11 +787,17 @@ ReadBuilder constantFieldIds(Set<Integer> newConstantFieldIds) {
787787

788788
public <D> CloseableIterable<D> build() {
789789
Preconditions.checkNotNull(schema, "Schema is required");
790+
Set<Integer> idsToExclude =
791+
Sets.difference(
792+
Sets.union(constantFieldIds, MetadataColumns.metadataFieldIds()),
793+
ImmutableSet.of(
794+
MetadataColumns.ROW_ID.fieldId(),
795+
MetadataColumns.LAST_UPDATED_SEQUENCE_NUMBER.fieldId()));
796+
790797
return new OrcIterable<>(
791798
file,
792799
conf,
793-
TypeUtil.selectNot(
794-
schema, Sets.union(constantFieldIds, MetadataColumns.metadataFieldIds())),
800+
TypeUtil.selectNot(schema, idsToExclude),
795801
nameMapping,
796802
start,
797803
length,

0 commit comments

Comments
 (0)