You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ao...@apache.org on 2023/12/12 09:17:14 UTC
(iceberg) branch main updated: Data: Add GenericFileWriterFactory (#9267)
This is an automated email from the ASF dual-hosted git repository.
aokolnychyi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 0c5b87a27e Data: Add GenericFileWriterFactory (#9267)
0c5b87a27e is described below
commit 0c5b87a27e0564ddaae2fa1cb93c32dd7a5e8681
Author: Anton Okolnychyi <ao...@apple.com>
AuthorDate: Tue Dec 12 10:17:08 2023 +0100
Data: Add GenericFileWriterFactory (#9267)
---
.../iceberg/data/GenericFileWriterFactory.java | 196 +++++++++++++++++++++
.../java/org/apache/iceberg/data/FileHelpers.java | 77 +++-----
.../iceberg/data/TestGenericFileWriterFactory.java | 65 +++++++
.../apache/iceberg/io/TestFileWriterFactory.java | 9 +-
.../source/TestMetadataTableReadableMetrics.java | 20 +--
.../source/TestMetadataTableReadableMetrics.java | 20 +--
.../source/TestMetadataTableReadableMetrics.java | 20 +--
7 files changed, 319 insertions(+), 88 deletions(-)
diff --git a/data/src/main/java/org/apache/iceberg/data/GenericFileWriterFactory.java b/data/src/main/java/org/apache/iceberg/data/GenericFileWriterFactory.java
new file mode 100644
index 0000000000..4e55743442
--- /dev/null
+++ b/data/src/main/java/org/apache/iceberg/data/GenericFileWriterFactory.java
@@ -0,0 +1,196 @@
+/*
+ * 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.iceberg.data;
+
+import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT;
+import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT_DEFAULT;
+import static org.apache.iceberg.TableProperties.DELETE_DEFAULT_FILE_FORMAT;
+
+import java.util.Map;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.data.avro.DataWriter;
+import org.apache.iceberg.data.orc.GenericOrcWriter;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.orc.ORC;
+import org.apache.iceberg.parquet.Parquet;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+class GenericFileWriterFactory extends BaseFileWriterFactory<Record> {
+
+ GenericFileWriterFactory(
+ Table table,
+ FileFormat dataFileFormat,
+ Schema dataSchema,
+ SortOrder dataSortOrder,
+ FileFormat deleteFileFormat,
+ int[] equalityFieldIds,
+ Schema equalityDeleteRowSchema,
+ SortOrder equalityDeleteSortOrder,
+ Schema positionDeleteRowSchema) {
+ super(
+ table,
+ dataFileFormat,
+ dataSchema,
+ dataSortOrder,
+ deleteFileFormat,
+ equalityFieldIds,
+ equalityDeleteRowSchema,
+ equalityDeleteSortOrder,
+ positionDeleteRowSchema);
+ }
+
+ static Builder builderFor(Table table) {
+ return new Builder(table);
+ }
+
+ @Override
+ protected void configureDataWrite(Avro.DataWriteBuilder builder) {
+ builder.createWriterFunc(DataWriter::create);
+ }
+
+ @Override
+ protected void configureEqualityDelete(Avro.DeleteWriteBuilder builder) {
+ builder.createWriterFunc(DataWriter::create);
+ }
+
+ @Override
+ protected void configurePositionDelete(Avro.DeleteWriteBuilder builder) {
+ builder.createWriterFunc(DataWriter::create);
+ }
+
+ @Override
+ protected void configureDataWrite(Parquet.DataWriteBuilder builder) {
+ builder.createWriterFunc(GenericParquetWriter::buildWriter);
+ }
+
+ @Override
+ protected void configureEqualityDelete(Parquet.DeleteWriteBuilder builder) {
+ builder.createWriterFunc(GenericParquetWriter::buildWriter);
+ }
+
+ @Override
+ protected void configurePositionDelete(Parquet.DeleteWriteBuilder builder) {
+ builder.createWriterFunc(GenericParquetWriter::buildWriter);
+ }
+
+ @Override
+ protected void configureDataWrite(ORC.DataWriteBuilder builder) {
+ builder.createWriterFunc(GenericOrcWriter::buildWriter);
+ }
+
+ @Override
+ protected void configureEqualityDelete(ORC.DeleteWriteBuilder builder) {
+ builder.createWriterFunc(GenericOrcWriter::buildWriter);
+ }
+
+ @Override
+ protected void configurePositionDelete(ORC.DeleteWriteBuilder builder) {
+ builder.createWriterFunc(GenericOrcWriter::buildWriter);
+ }
+
+ static class Builder {
+ private final Table table;
+ private FileFormat dataFileFormat;
+ private Schema dataSchema;
+ private SortOrder dataSortOrder;
+ private FileFormat deleteFileFormat;
+ private int[] equalityFieldIds;
+ private Schema equalityDeleteRowSchema;
+ private SortOrder equalityDeleteSortOrder;
+ private Schema positionDeleteRowSchema;
+
+ Builder(Table table) {
+ this.table = table;
+ this.dataSchema = table.schema();
+
+ Map<String, String> properties = table.properties();
+
+ String dataFileFormatName =
+ properties.getOrDefault(DEFAULT_FILE_FORMAT, DEFAULT_FILE_FORMAT_DEFAULT);
+ this.dataFileFormat = FileFormat.fromString(dataFileFormatName);
+
+ String deleteFileFormatName =
+ properties.getOrDefault(DELETE_DEFAULT_FILE_FORMAT, dataFileFormatName);
+ this.deleteFileFormat = FileFormat.fromString(deleteFileFormatName);
+ }
+
+ Builder dataFileFormat(FileFormat newDataFileFormat) {
+ this.dataFileFormat = newDataFileFormat;
+ return this;
+ }
+
+ Builder dataSchema(Schema newDataSchema) {
+ this.dataSchema = newDataSchema;
+ return this;
+ }
+
+ Builder dataSortOrder(SortOrder newDataSortOrder) {
+ this.dataSortOrder = newDataSortOrder;
+ return this;
+ }
+
+ Builder deleteFileFormat(FileFormat newDeleteFileFormat) {
+ this.deleteFileFormat = newDeleteFileFormat;
+ return this;
+ }
+
+ Builder equalityFieldIds(int[] newEqualityFieldIds) {
+ this.equalityFieldIds = newEqualityFieldIds;
+ return this;
+ }
+
+ Builder equalityDeleteRowSchema(Schema newEqualityDeleteRowSchema) {
+ this.equalityDeleteRowSchema = newEqualityDeleteRowSchema;
+ return this;
+ }
+
+ Builder equalityDeleteSortOrder(SortOrder newEqualityDeleteSortOrder) {
+ this.equalityDeleteSortOrder = newEqualityDeleteSortOrder;
+ return this;
+ }
+
+ Builder positionDeleteRowSchema(Schema newPositionDeleteRowSchema) {
+ this.positionDeleteRowSchema = newPositionDeleteRowSchema;
+ return this;
+ }
+
+ GenericFileWriterFactory build() {
+ boolean noEqualityDeleteConf = equalityFieldIds == null && equalityDeleteRowSchema == null;
+ boolean fullEqualityDeleteConf = equalityFieldIds != null && equalityDeleteRowSchema != null;
+ Preconditions.checkArgument(
+ noEqualityDeleteConf || fullEqualityDeleteConf,
+ "Equality field IDs and equality delete row schema must be set together");
+
+ return new GenericFileWriterFactory(
+ table,
+ dataFileFormat,
+ dataSchema,
+ dataSortOrder,
+ deleteFileFormat,
+ equalityFieldIds,
+ equalityDeleteRowSchema,
+ equalityDeleteSortOrder,
+ positionDeleteRowSchema);
+ }
+ }
+}
diff --git a/data/src/test/java/org/apache/iceberg/data/FileHelpers.java b/data/src/test/java/org/apache/iceberg/data/FileHelpers.java
index 557e775057..62df1634aa 100644
--- a/data/src/test/java/org/apache/iceberg/data/FileHelpers.java
+++ b/data/src/test/java/org/apache/iceberg/data/FileHelpers.java
@@ -18,17 +18,11 @@
*/
package org.apache.iceberg.data;
-import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT;
-import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT_DEFAULT;
-
import java.io.Closeable;
import java.io.IOException;
import java.util.List;
-import java.util.Map;
import org.apache.iceberg.DataFile;
-import org.apache.iceberg.DataFiles;
import org.apache.iceberg.DeleteFile;
-import org.apache.iceberg.FileFormat;
import org.apache.iceberg.Schema;
import org.apache.iceberg.StructLike;
import org.apache.iceberg.Table;
@@ -38,8 +32,8 @@ import org.apache.iceberg.deletes.PositionDeleteWriter;
import org.apache.iceberg.encryption.EncryptedFiles;
import org.apache.iceberg.encryption.EncryptedOutputFile;
import org.apache.iceberg.encryption.EncryptionKeyMetadata;
-import org.apache.iceberg.io.FileAppender;
-import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.DataWriter;
+import org.apache.iceberg.io.FileWriterFactory;
import org.apache.iceberg.io.OutputFile;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.util.CharSequenceSet;
@@ -56,11 +50,10 @@ public class FileHelpers {
public static Pair<DeleteFile, CharSequenceSet> writeDeleteFile(
Table table, OutputFile out, StructLike partition, List<Pair<CharSequence, Long>> deletes)
throws IOException {
- FileFormat format = defaultFormat(table.properties());
- FileAppenderFactory<Record> factory = new GenericAppenderFactory(table.schema(), table.spec());
+ FileWriterFactory<Record> factory = GenericFileWriterFactory.builderFor(table).build();
PositionDeleteWriter<Record> writer =
- factory.newPosDeleteWriter(encrypt(out), format, partition);
+ factory.newPositionDeleteWriter(encrypt(out), table.spec(), partition);
PositionDelete<Record> posDelete = PositionDelete.create();
try (Closeable toClose = writer) {
for (Pair<CharSequence, Long> delete : deletes) {
@@ -84,15 +77,16 @@ public class FileHelpers {
List<Record> deletes,
Schema deleteRowSchema)
throws IOException {
- FileFormat format = defaultFormat(table.properties());
int[] equalityFieldIds =
deleteRowSchema.columns().stream().mapToInt(Types.NestedField::fieldId).toArray();
- FileAppenderFactory<Record> factory =
- new GenericAppenderFactory(
- table.schema(), table.spec(), equalityFieldIds, deleteRowSchema, null);
+ FileWriterFactory<Record> factory =
+ GenericFileWriterFactory.builderFor(table)
+ .equalityDeleteRowSchema(deleteRowSchema)
+ .equalityFieldIds(equalityFieldIds)
+ .build();
EqualityDeleteWriter<Record> writer =
- factory.newEqDeleteWriter(encrypt(out), format, partition);
+ factory.newEqualityDeleteWriter(encrypt(out), table.spec(), partition);
try (Closeable toClose = writer) {
writer.write(deletes);
}
@@ -102,56 +96,36 @@ public class FileHelpers {
public static DataFile writeDataFile(Table table, OutputFile out, List<Record> rows)
throws IOException {
- FileFormat format = defaultFormat(table.properties());
- GenericAppenderFactory factory = new GenericAppenderFactory(table.schema());
+ FileWriterFactory<Record> factory = GenericFileWriterFactory.builderFor(table).build();
- FileAppender<Record> writer = factory.newAppender(out, format);
+ DataWriter<Record> writer = factory.newDataWriter(encrypt(out), table.spec(), null);
try (Closeable toClose = writer) {
- writer.addAll(rows);
+ writer.write(rows);
}
- return DataFiles.builder(table.spec())
- .withFormat(format)
- .withPath(out.location())
- .withFileSizeInBytes(writer.length())
- .withSplitOffsets(writer.splitOffsets())
- .withMetrics(writer.metrics())
- .build();
+ return writer.toDataFile();
}
public static DataFile writeDataFile(
Table table, OutputFile out, StructLike partition, List<Record> rows) throws IOException {
- FileFormat format = defaultFormat(table.properties());
- GenericAppenderFactory factory = new GenericAppenderFactory(table.schema(), table.spec());
+ FileWriterFactory<Record> factory = GenericFileWriterFactory.builderFor(table).build();
- FileAppender<Record> writer = factory.newAppender(out, format);
+ DataWriter<Record> writer = factory.newDataWriter(encrypt(out), table.spec(), partition);
try (Closeable toClose = writer) {
- writer.addAll(rows);
+ writer.write(rows);
}
- return DataFiles.builder(table.spec())
- .withFormat(format)
- .withPath(out.location())
- .withPartition(partition)
- .withFileSizeInBytes(writer.length())
- .withSplitOffsets(writer.splitOffsets())
- .withMetrics(writer.metrics())
- .build();
+ return writer.toDataFile();
}
public static DeleteFile writePosDeleteFile(
Table table, OutputFile out, StructLike partition, List<PositionDelete<?>> deletes)
throws IOException {
- FileFormat format = defaultFormat(table.properties());
- FileAppenderFactory<Record> factory =
- new GenericAppenderFactory(
- table.schema(),
- table.spec(),
- null, // Equality Fields
- null, // Equality Delete row schema
- table.schema()); // Position Delete row schema (will be wrapped)
-
- PositionDeleteWriter<?> writer = factory.newPosDeleteWriter(encrypt(out), format, partition);
+ FileWriterFactory<Record> factory =
+ GenericFileWriterFactory.builderFor(table).positionDeleteRowSchema(table.schema()).build();
+
+ PositionDeleteWriter<?> writer =
+ factory.newPositionDeleteWriter(encrypt(out), table.spec(), partition);
try (Closeable toClose = writer) {
for (PositionDelete delete : deletes) {
writer.write(delete);
@@ -164,9 +138,4 @@ public class FileHelpers {
private static EncryptedOutputFile encrypt(OutputFile out) {
return EncryptedFiles.encryptedOutput(out, EncryptionKeyMetadata.EMPTY);
}
-
- private static FileFormat defaultFormat(Map<String, String> properties) {
- String formatString = properties.getOrDefault(DEFAULT_FILE_FORMAT, DEFAULT_FILE_FORMAT_DEFAULT);
- return FileFormat.fromString(formatString);
- }
}
diff --git a/data/src/test/java/org/apache/iceberg/data/TestGenericFileWriterFactory.java b/data/src/test/java/org/apache/iceberg/data/TestGenericFileWriterFactory.java
new file mode 100644
index 0000000000..6752234f5e
--- /dev/null
+++ b/data/src/test/java/org/apache/iceberg/data/TestGenericFileWriterFactory.java
@@ -0,0 +1,65 @@
+/*
+ * 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.iceberg.data;
+
+import java.util.List;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.io.FileWriterFactory;
+import org.apache.iceberg.io.TestFileWriterFactory;
+import org.apache.iceberg.util.ArrayUtil;
+import org.apache.iceberg.util.StructLikeSet;
+
+public class TestGenericFileWriterFactory extends TestFileWriterFactory<Record> {
+
+ public TestGenericFileWriterFactory(FileFormat fileFormat, boolean partitioned) {
+ super(fileFormat, partitioned);
+ }
+
+ @Override
+ protected FileWriterFactory<Record> newWriterFactory(
+ Schema dataSchema,
+ List<Integer> equalityFieldIds,
+ Schema equalityDeleteRowSchema,
+ Schema positionDeleteRowSchema) {
+ return GenericFileWriterFactory.builderFor(table)
+ .dataSchema(dataSchema)
+ .dataFileFormat(format())
+ .deleteFileFormat(format())
+ .equalityFieldIds(ArrayUtil.toIntArray(equalityFieldIds))
+ .equalityDeleteRowSchema(equalityDeleteRowSchema)
+ .positionDeleteRowSchema(positionDeleteRowSchema)
+ .build();
+ }
+
+ @Override
+ protected Record toRow(Integer id, String data) {
+ GenericRecord record = GenericRecord.create(table.schema().asStruct());
+ record.set(0, id);
+ record.set(1, data);
+ return record;
+ }
+
+ @Override
+ protected StructLikeSet toSet(Iterable<Record> records) {
+ StructLikeSet set = StructLikeSet.create(table.schema().asStruct());
+ records.forEach(set::add);
+ return set;
+ }
+}
diff --git a/data/src/test/java/org/apache/iceberg/io/TestFileWriterFactory.java b/data/src/test/java/org/apache/iceberg/io/TestFileWriterFactory.java
index e25a179edb..152679abbd 100644
--- a/data/src/test/java/org/apache/iceberg/io/TestFileWriterFactory.java
+++ b/data/src/test/java/org/apache/iceberg/io/TestFileWriterFactory.java
@@ -76,18 +76,15 @@ public abstract class TestFileWriterFactory<T> extends WriterTestBase<T> {
private final FileFormat fileFormat;
private final boolean partitioned;
- private final List<T> dataRows;
private StructLike partition = null;
private OutputFileFactory fileFactory = null;
+ private List<T> dataRows;
public TestFileWriterFactory(FileFormat fileFormat, boolean partitioned) {
super(TABLE_FORMAT_VERSION);
this.fileFormat = fileFormat;
this.partitioned = partitioned;
- this.dataRows =
- ImmutableList.of(
- toRow(1, "aaa"), toRow(2, "aaa"), toRow(3, "aaa"), toRow(4, "aaa"), toRow(5, "aaa"));
}
protected abstract StructLikeSet toSet(Iterable<T> records);
@@ -113,6 +110,10 @@ public abstract class TestFileWriterFactory<T> extends WriterTestBase<T> {
}
this.fileFactory = OutputFileFactory.builderFor(table, 1, 1).format(fileFormat).build();
+
+ this.dataRows =
+ ImmutableList.of(
+ toRow(1, "aaa"), toRow(2, "aaa"), toRow(3, "aaa"), toRow(4, "aaa"), toRow(5, "aaa"));
}
@Test
diff --git a/flink/v1.16/flink/src/test/java/org/apache/iceberg/flink/source/TestMetadataTableReadableMetrics.java b/flink/v1.16/flink/src/test/java/org/apache/iceberg/flink/source/TestMetadataTableReadableMetrics.java
index cb6fda18a1..f05bf2fcd9 100644
--- a/flink/v1.16/flink/src/test/java/org/apache/iceberg/flink/source/TestMetadataTableReadableMetrics.java
+++ b/flink/v1.16/flink/src/test/java/org/apache/iceberg/flink/source/TestMetadataTableReadableMetrics.java
@@ -219,27 +219,27 @@ public class TestMetadataTableReadableMetrics extends FlinkCatalogTestBase {
Row binaryCol =
Row.of(
- 59L,
+ 52L,
4L,
2L,
null,
Base64.getDecoder().decode("1111"),
Base64.getDecoder().decode("2222"));
- Row booleanCol = Row.of(44L, 4L, 0L, null, false, true);
- Row decimalCol = Row.of(97L, 4L, 1L, null, new BigDecimal("1.00"), new BigDecimal("2.00"));
- Row doubleCol = Row.of(99L, 4L, 0L, 1L, 1.0D, 2.0D);
+ Row booleanCol = Row.of(32L, 4L, 0L, null, false, true);
+ Row decimalCol = Row.of(85L, 4L, 1L, null, new BigDecimal("1.00"), new BigDecimal("2.00"));
+ Row doubleCol = Row.of(85L, 4L, 0L, 1L, 1.0D, 2.0D);
Row fixedCol =
Row.of(
- 55L,
+ 44L,
4L,
2L,
null,
Base64.getDecoder().decode("1111"),
Base64.getDecoder().decode("2222"));
- Row floatCol = Row.of(90L, 4L, 0L, 2L, 0f, 0f);
- Row intCol = Row.of(91L, 4L, 0L, null, 1, 2);
- Row longCol = Row.of(91L, 4L, 0L, null, 1L, 2L);
- Row stringCol = Row.of(99L, 4L, 0L, null, "1", "2");
+ Row floatCol = Row.of(71L, 4L, 0L, 2L, 0f, 0f);
+ Row intCol = Row.of(71L, 4L, 0L, null, 1, 2);
+ Row longCol = Row.of(79L, 4L, 0L, null, 1L, 2L);
+ Row stringCol = Row.of(79L, 4L, 0L, null, "1", "2");
List<Row> expected =
Lists.newArrayList(
@@ -291,7 +291,7 @@ public class TestMetadataTableReadableMetrics extends FlinkCatalogTestBase {
public void testNestedValues() throws Exception {
createNestedTable();
- Row leafDoubleCol = Row.of(53L, 3L, 1L, 1L, 0.0D, 0.0D);
+ Row leafDoubleCol = Row.of(46L, 3L, 1L, 1L, 0.0D, 0.0D);
Row leafLongCol = Row.of(54L, 3L, 1L, null, 0L, 1L);
Row metrics = Row.of(Row.of(leafDoubleCol, leafLongCol));
diff --git a/flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/source/TestMetadataTableReadableMetrics.java b/flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/source/TestMetadataTableReadableMetrics.java
index cb6fda18a1..f05bf2fcd9 100644
--- a/flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/source/TestMetadataTableReadableMetrics.java
+++ b/flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/source/TestMetadataTableReadableMetrics.java
@@ -219,27 +219,27 @@ public class TestMetadataTableReadableMetrics extends FlinkCatalogTestBase {
Row binaryCol =
Row.of(
- 59L,
+ 52L,
4L,
2L,
null,
Base64.getDecoder().decode("1111"),
Base64.getDecoder().decode("2222"));
- Row booleanCol = Row.of(44L, 4L, 0L, null, false, true);
- Row decimalCol = Row.of(97L, 4L, 1L, null, new BigDecimal("1.00"), new BigDecimal("2.00"));
- Row doubleCol = Row.of(99L, 4L, 0L, 1L, 1.0D, 2.0D);
+ Row booleanCol = Row.of(32L, 4L, 0L, null, false, true);
+ Row decimalCol = Row.of(85L, 4L, 1L, null, new BigDecimal("1.00"), new BigDecimal("2.00"));
+ Row doubleCol = Row.of(85L, 4L, 0L, 1L, 1.0D, 2.0D);
Row fixedCol =
Row.of(
- 55L,
+ 44L,
4L,
2L,
null,
Base64.getDecoder().decode("1111"),
Base64.getDecoder().decode("2222"));
- Row floatCol = Row.of(90L, 4L, 0L, 2L, 0f, 0f);
- Row intCol = Row.of(91L, 4L, 0L, null, 1, 2);
- Row longCol = Row.of(91L, 4L, 0L, null, 1L, 2L);
- Row stringCol = Row.of(99L, 4L, 0L, null, "1", "2");
+ Row floatCol = Row.of(71L, 4L, 0L, 2L, 0f, 0f);
+ Row intCol = Row.of(71L, 4L, 0L, null, 1, 2);
+ Row longCol = Row.of(79L, 4L, 0L, null, 1L, 2L);
+ Row stringCol = Row.of(79L, 4L, 0L, null, "1", "2");
List<Row> expected =
Lists.newArrayList(
@@ -291,7 +291,7 @@ public class TestMetadataTableReadableMetrics extends FlinkCatalogTestBase {
public void testNestedValues() throws Exception {
createNestedTable();
- Row leafDoubleCol = Row.of(53L, 3L, 1L, 1L, 0.0D, 0.0D);
+ Row leafDoubleCol = Row.of(46L, 3L, 1L, 1L, 0.0D, 0.0D);
Row leafLongCol = Row.of(54L, 3L, 1L, null, 0L, 1L);
Row metrics = Row.of(Row.of(leafDoubleCol, leafLongCol));
diff --git a/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/source/TestMetadataTableReadableMetrics.java b/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/source/TestMetadataTableReadableMetrics.java
index cb6fda18a1..f05bf2fcd9 100644
--- a/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/source/TestMetadataTableReadableMetrics.java
+++ b/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/source/TestMetadataTableReadableMetrics.java
@@ -219,27 +219,27 @@ public class TestMetadataTableReadableMetrics extends FlinkCatalogTestBase {
Row binaryCol =
Row.of(
- 59L,
+ 52L,
4L,
2L,
null,
Base64.getDecoder().decode("1111"),
Base64.getDecoder().decode("2222"));
- Row booleanCol = Row.of(44L, 4L, 0L, null, false, true);
- Row decimalCol = Row.of(97L, 4L, 1L, null, new BigDecimal("1.00"), new BigDecimal("2.00"));
- Row doubleCol = Row.of(99L, 4L, 0L, 1L, 1.0D, 2.0D);
+ Row booleanCol = Row.of(32L, 4L, 0L, null, false, true);
+ Row decimalCol = Row.of(85L, 4L, 1L, null, new BigDecimal("1.00"), new BigDecimal("2.00"));
+ Row doubleCol = Row.of(85L, 4L, 0L, 1L, 1.0D, 2.0D);
Row fixedCol =
Row.of(
- 55L,
+ 44L,
4L,
2L,
null,
Base64.getDecoder().decode("1111"),
Base64.getDecoder().decode("2222"));
- Row floatCol = Row.of(90L, 4L, 0L, 2L, 0f, 0f);
- Row intCol = Row.of(91L, 4L, 0L, null, 1, 2);
- Row longCol = Row.of(91L, 4L, 0L, null, 1L, 2L);
- Row stringCol = Row.of(99L, 4L, 0L, null, "1", "2");
+ Row floatCol = Row.of(71L, 4L, 0L, 2L, 0f, 0f);
+ Row intCol = Row.of(71L, 4L, 0L, null, 1, 2);
+ Row longCol = Row.of(79L, 4L, 0L, null, 1L, 2L);
+ Row stringCol = Row.of(79L, 4L, 0L, null, "1", "2");
List<Row> expected =
Lists.newArrayList(
@@ -291,7 +291,7 @@ public class TestMetadataTableReadableMetrics extends FlinkCatalogTestBase {
public void testNestedValues() throws Exception {
createNestedTable();
- Row leafDoubleCol = Row.of(53L, 3L, 1L, 1L, 0.0D, 0.0D);
+ Row leafDoubleCol = Row.of(46L, 3L, 1L, 1L, 0.0D, 0.0D);
Row leafLongCol = Row.of(54L, 3L, 1L, null, 0L, 1L);
Row metrics = Row.of(Row.of(leafDoubleCol, leafLongCol));