You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "rdblue (via GitHub)" <gi...@apache.org> on 2025/03/06 19:35:06 UTC
[PR] Flink: Support Avro timestamp(9), unknown, and defaults [iceberg]
rdblue opened a new pull request, #12470:
URL: https://github.com/apache/iceberg/pull/12470
This updates Flink's Avro readers to support new timestamp(9) and unknown types.
While enabling `DataTest` cases, I found that `supportsDefaultValues` was not enabled so default value tests were not running for Avro. After I enabled those tests, I also needed to update the `RowData` assertions and also convert values to match Flink's object model in the readers by calling `RowDataUtil.convertConstant`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.20: Support Avro and Parquet timestamp(9), unknown, and defaults [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12470:
URL: https://github.com/apache/iceberg/pull/12470#discussion_r1995433373
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestHelpers.java:
##########
@@ -187,11 +188,38 @@ public static void assertRowData(
types.add(field.type());
}
- for (int i = 0; i < types.size(); i += 1) {
- LogicalType logicalType = ((RowType) rowType).getTypeAt(i);
- Object expected = expectedRecord.get(i, Object.class);
- Object actual = FlinkRowData.createFieldGetter(logicalType, i).getFieldOrNull(actualRowData);
- assertEquals(types.get(i), logicalType, expected, actual);
+ if (expectedRecord instanceof Record) {
+ Record expected = (Record) expectedRecord;
+ Types.StructType expectedType = expected.struct();
+ int pos = 0;
+ for (Types.NestedField field : structType.fields()) {
+ Types.NestedField expectedField = expectedType.field(field.fieldId());
+ LogicalType logicalType = ((RowType) rowType).getTypeAt(pos);
+ Object actualValue =
+ FlinkRowData.createFieldGetter(logicalType, pos).getFieldOrNull(actualRowData);
+ if (expectedField != null) {
+ assertEquals(
+ field.type(), logicalType, expected.getField(expectedField.name()), actualValue);
+ } else {
+ // convert the initial value to generic because that is the data model used to generate
+ // the expected records
+ assertEquals(
+ field.type(),
+ logicalType,
+ GenericDataUtil.internalToGeneric(field.type(), field.initialDefault()),
+ actualValue);
+ }
+ pos += 1;
+ }
+
Review Comment:
nit: this is a bit strange. Could we fix the empty lines?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.20: Support Avro and Parquet timestamp(9), unknown, and defaults [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12470:
URL: https://github.com/apache/iceberg/pull/12470#discussion_r1995432725
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkAvroReaderWriter.java:
##########
@@ -0,0 +1,142 @@
+/*
+ * 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.flink.data;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.List;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.iceberg.Files;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.data.DataTest;
+import org.apache.iceberg.data.RandomGenericData;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.avro.DataWriter;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.flink.TestHelpers;
+import org.apache.iceberg.inmemory.InMemoryOutputFile;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+
+public class TestFlinkAvroReaderWriter extends DataTest {
+
+ private static final int NUM_RECORDS = 100;
+
+ @Override
+ protected boolean supportsDefaultValues() {
+ return true;
+ }
+
+ @Override
+ protected boolean supportsUnknown() {
+ return true;
+ }
+
+ @Override
+ protected boolean supportsTimestampNanos() {
+ return true;
+ }
+
+ @Override
+ protected void writeAndValidate(Schema schema) throws IOException {
+ List<Record> expectedRecords = RandomGenericData.generate(schema, NUM_RECORDS, 1991L);
+ writeAndValidate(schema, expectedRecords);
+ }
+
+ @Override
+ protected void writeAndValidate(Schema schema, List<Record> expectedRecords) throws IOException {
+ writeAndValidate(schema, schema, expectedRecords);
+ }
+
+ @Override
+ protected void writeAndValidate(Schema writeSchema, Schema expectedSchema) throws IOException {
+ List<Record> expectedRecords = RandomGenericData.generate(writeSchema, NUM_RECORDS, 1991L);
+ writeAndValidate(writeSchema, expectedSchema, expectedRecords);
+ }
+
+ protected void writeAndValidate(
+ Schema writeSchema, Schema expectedSchema, List<Record> expectedRecords) throws IOException {
+ List<RowData> expectedRows =
+ Lists.newArrayList(RandomRowData.convert(writeSchema, expectedRecords));
+
+ File recordsFile = File.createTempFile("junit", null, temp.toFile());
Review Comment:
Could we use `OutputFile outputFile = new InMemoryOutputFile();` here too?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support Avro and Parquet timestamp(9), unknown, and defaults [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #12470:
URL: https://github.com/apache/iceberg/pull/12470#discussion_r1984040406
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetWriters.java:
##########
@@ -134,43 +145,15 @@ private ParquetValueWriter<?> newOption(Type fieldType, ParquetValueWriter<?> wr
public ParquetValueWriter<?> primitive(LogicalType fType, PrimitiveType primitive) {
ColumnDescriptor desc = type.getColumnDescription(currentPath());
- if (primitive.getOriginalType() != null) {
- switch (primitive.getOriginalType()) {
- case ENUM:
- case JSON:
- case UTF8:
- return strings(desc);
- case DATE:
- case INT_8:
- case INT_16:
- case INT_32:
- return ints(fType, desc);
- case INT_64:
- return ParquetValueWriters.longs(desc);
- case TIME_MICROS:
- return timeMicros(desc);
- case TIMESTAMP_MICROS:
- return timestamps(desc);
- case DECIMAL:
- DecimalLogicalTypeAnnotation decimal =
- (DecimalLogicalTypeAnnotation) primitive.getLogicalTypeAnnotation();
- switch (primitive.getPrimitiveTypeName()) {
- case INT32:
- return decimalAsInteger(desc, decimal.getPrecision(), decimal.getScale());
- case INT64:
- return decimalAsLong(desc, decimal.getPrecision(), decimal.getScale());
- case BINARY:
- case FIXED_LEN_BYTE_ARRAY:
- return decimalAsFixed(desc, decimal.getPrecision(), decimal.getScale());
- default:
- throw new UnsupportedOperationException(
- "Unsupported base type for decimal: " + primitive.getPrimitiveTypeName());
- }
- case BSON:
- return byteArrays(desc);
- default:
- throw new UnsupportedOperationException(
- "Unsupported logical type: " + primitive.getOriginalType());
+ LogicalTypeAnnotation annotation = primitive.getLogicalTypeAnnotation();
+ if (annotation != null) {
+ Optional<ParquetValueWriter<?>> writer =
+ annotation.accept(new LogicalTypeWriterBuilder(fType, desc));
Review Comment:
Updated this to use the logical annotation visitor.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.20: Support Avro and Parquet timestamp(9), unknown, and defaults [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #12470:
URL: https://github.com/apache/iceberg/pull/12470#discussion_r1996274844
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestHelpers.java:
##########
@@ -187,11 +188,38 @@ public static void assertRowData(
types.add(field.type());
}
- for (int i = 0; i < types.size(); i += 1) {
- LogicalType logicalType = ((RowType) rowType).getTypeAt(i);
- Object expected = expectedRecord.get(i, Object.class);
- Object actual = FlinkRowData.createFieldGetter(logicalType, i).getFieldOrNull(actualRowData);
- assertEquals(types.get(i), logicalType, expected, actual);
+ if (expectedRecord instanceof Record) {
+ Record expected = (Record) expectedRecord;
+ Types.StructType expectedType = expected.struct();
+ int pos = 0;
+ for (Types.NestedField field : structType.fields()) {
+ Types.NestedField expectedField = expectedType.field(field.fieldId());
+ LogicalType logicalType = ((RowType) rowType).getTypeAt(pos);
+ Object actualValue =
+ FlinkRowData.createFieldGetter(logicalType, pos).getFieldOrNull(actualRowData);
+ if (expectedField != null) {
+ assertEquals(
+ field.type(), logicalType, expected.getField(expectedField.name()), actualValue);
+ } else {
+ // convert the initial value to generic because that is the data model used to generate
+ // the expected records
+ assertEquals(
+ field.type(),
+ logicalType,
+ GenericDataUtil.internalToGeneric(field.type(), field.initialDefault()),
+ actualValue);
+ }
+ pos += 1;
+ }
+
Review Comment:
For dense control flow blocks, we often leave extra newlines to make it more readable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.20: Support Avro and Parquet timestamp(9), unknown, and defaults [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #12470:
URL: https://github.com/apache/iceberg/pull/12470#discussion_r1996281732
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetWriters.java:
##########
@@ -194,8 +177,104 @@ public ParquetValueWriter<?> primitive(LogicalType fType, PrimitiveType primitiv
}
}
- private static ParquetValueWriters.PrimitiveWriter<?> ints(
- LogicalType type, ColumnDescriptor desc) {
+ private static class LogicalTypeWriterBuilder
+ implements LogicalTypeAnnotationVisitor<ParquetValueWriter<?>> {
+ private final LogicalType flinkType;
+ private final ColumnDescriptor desc;
+
+ private LogicalTypeWriterBuilder(LogicalType flinkType, ColumnDescriptor desc) {
+ this.flinkType = flinkType;
+ this.desc = desc;
+ }
+
+ @Override
+ public Optional<ParquetValueWriter<?>> visit(StringLogicalTypeAnnotation strings) {
+ return Optional.of(strings(desc));
+ }
+
+ @Override
+ public Optional<ParquetValueWriter<?>> visit(EnumLogicalTypeAnnotation enums) {
+ return Optional.of(strings(desc));
+ }
+
+ @Override
+ public Optional<ParquetValueWriter<?>> visit(DecimalLogicalTypeAnnotation decimal) {
+ ParquetValueWriter<DecimalData> writer;
+ switch (desc.getPrimitiveType().getPrimitiveTypeName()) {
+ case INT32:
+ writer = decimalAsInteger(desc, decimal.getPrecision(), decimal.getScale());
+ break;
+ case INT64:
+ writer = decimalAsLong(desc, decimal.getPrecision(), decimal.getScale());
+ break;
+ case BINARY:
+ case FIXED_LEN_BYTE_ARRAY:
+ writer = decimalAsFixed(desc, decimal.getPrecision(), decimal.getScale());
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ "Unsupported base type for decimal: "
+ + desc.getPrimitiveType().getPrimitiveTypeName());
+ }
+ return Optional.of(writer);
+ }
+
+ @Override
+ public Optional<ParquetValueWriter<?>> visit(DateLogicalTypeAnnotation dates) {
+ return Optional.of(ints(flinkType, desc));
+ }
+
+ @Override
+ public Optional<ParquetValueWriter<?>> visit(TimeLogicalTypeAnnotation times) {
+ Preconditions.checkArgument(
+ LogicalTypeAnnotation.TimeUnit.MICROS.equals(times.getUnit()),
+ "Cannot write time in %s, only MICROS is supported",
+ times.getUnit());
+ return Optional.of(timeMicros(desc));
+ }
+
+ @Override
+ public Optional<ParquetValueWriter<?>> visit(TimestampLogicalTypeAnnotation timestamps) {
+ ParquetValueWriter<TimestampData> writer;
+ switch (timestamps.getUnit()) {
+ case NANOS:
+ writer = timestampNanos(desc);
+ break;
+ case MICROS:
+ writer = timestamps(desc);
+ break;
+ default:
+ throw new UnsupportedOperationException("Unsupported timestamp type: " + timestamps);
+ }
+
+ return Optional.of(writer);
+ }
+
+ @Override
+ public Optional<ParquetValueWriter<?>> visit(IntLogicalTypeAnnotation type) {
+ Preconditions.checkArgument(type.isSigned(), "Cannot write signed integer type: %s", type);
Review Comment:
Yes. Good catch!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support Avro and Parquet timestamp(9), unknown, and defaults [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #12470:
URL: https://github.com/apache/iceberg/pull/12470#discussion_r1984039899
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetReaders.java:
##########
@@ -274,17 +272,11 @@ public Optional<ParquetValueReader<?>> visit(
public Optional<ParquetValueReader<?>> visit(
LogicalTypeAnnotation.TimestampLogicalTypeAnnotation timestampLogicalType) {
if (timestampLogicalType.getUnit() == LogicalTypeAnnotation.TimeUnit.MILLIS) {
- if (timestampLogicalType.isAdjustedToUTC()) {
- return Optional.of(new MillisToTimestampTzReader(desc));
- } else {
- return Optional.of(new MillisToTimestampReader(desc));
- }
+ return Optional.of(new MillisToTimestampReader(desc));
} else if (timestampLogicalType.getUnit() == LogicalTypeAnnotation.TimeUnit.MICROS) {
- if (timestampLogicalType.isAdjustedToUTC()) {
- return Optional.of(new MicrosToTimestampTzReader(desc));
- } else {
- return Optional.of(new MicrosToTimestampReader(desc));
- }
+ return Optional.of(new MicrosToTimestampReader(desc));
Review Comment:
Previously, the readers were converting values to `LocalDateTime` or `OffsetDateTime` and then Flink would convert those values back to a (`millis`, `nanosOfMilli`) pair. This involved a lot of unnecessary date/time logic in both Iceberg and Flink as well as readers to produce the separate types.
Now, the conversion to Flink is direct and doesn't go through Java date/time classes. That avoids all time zone calculations and should be quicker.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.20: Support Avro and Parquet timestamp(9), unknown, and defaults [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on PR #12470:
URL: https://github.com/apache/iceberg/pull/12470#issuecomment-2725892975
Thanks for reviewing, @pvary and @danielcweeks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support Avro timestamp(9), unknown, and defaults [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on PR #12470:
URL: https://github.com/apache/iceberg/pull/12470#issuecomment-2704775229
I'll also follow up with a PR for Parquet readers, but that depends on changes in #12463.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.20: Support Avro and Parquet timestamp(9), unknown, and defaults [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12470:
URL: https://github.com/apache/iceberg/pull/12470#discussion_r1995436870
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/data/AbstractTestFlinkAvroReaderWriter.java:
##########
@@ -1,182 +0,0 @@
-/*
- * 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.flink.data;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-import java.io.File;
-import java.io.IOException;
-import java.math.BigDecimal;
-import java.sql.Date;
-import java.sql.Time;
-import java.util.Iterator;
-import java.util.List;
-import org.apache.flink.table.data.RowData;
-import org.apache.flink.table.types.logical.RowType;
-import org.apache.iceberg.Files;
-import org.apache.iceberg.Schema;
-import org.apache.iceberg.avro.Avro;
-import org.apache.iceberg.data.DataTest;
-import org.apache.iceberg.data.GenericRecord;
-import org.apache.iceberg.data.RandomGenericData;
-import org.apache.iceberg.data.Record;
-import org.apache.iceberg.data.avro.DataReader;
-import org.apache.iceberg.data.avro.DataWriter;
-import org.apache.iceberg.flink.FlinkSchemaUtil;
-import org.apache.iceberg.flink.TestHelpers;
-import org.apache.iceberg.io.CloseableIterable;
-import org.apache.iceberg.io.FileAppender;
-import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
-import org.apache.iceberg.relocated.com.google.common.collect.Lists;
-import org.apache.iceberg.types.Types;
-import org.apache.iceberg.util.DateTimeUtil;
-import org.junit.jupiter.api.Test;
-
-public abstract class AbstractTestFlinkAvroReaderWriter extends DataTest {
-
- private static final int NUM_RECORDS = 100;
-
- private static final Schema SCHEMA_NUM_TYPE =
- new Schema(
- Types.NestedField.optional(1, "id", Types.IntegerType.get()),
- Types.NestedField.optional(2, "int", Types.IntegerType.get()),
- Types.NestedField.optional(3, "float", Types.FloatType.get()),
- Types.NestedField.optional(4, "double", Types.DoubleType.get()),
- Types.NestedField.optional(5, "date", Types.DateType.get()),
- Types.NestedField.optional(6, "time", Types.TimeType.get()),
- Types.NestedField.optional(7, "timestamp", Types.TimestampType.withoutZone()),
- Types.NestedField.optional(8, "bigint", Types.LongType.get()),
- Types.NestedField.optional(9, "decimal", Types.DecimalType.of(4, 2)));
-
- @Override
- protected void writeAndValidate(Schema schema) throws IOException {
- List<Record> expectedRecords = RandomGenericData.generate(schema, NUM_RECORDS, 1991L);
- writeAndValidate(schema, expectedRecords);
- }
-
- protected abstract Avro.ReadBuilder createAvroReadBuilder(File recordsFile, Schema schema);
-
- @Override
- protected void writeAndValidate(Schema schema, List<Record> expectedRecords) throws IOException {
- RowType flinkSchema = FlinkSchemaUtil.convert(schema);
- List<RowData> expectedRows = Lists.newArrayList(RandomRowData.convert(schema, expectedRecords));
-
- File recordsFile = File.createTempFile("junit", null, temp.toFile());
- assertThat(recordsFile.delete()).isTrue();
-
- // Write the expected records into AVRO file, then read them into RowData and assert with the
- // expected Record list.
- try (FileAppender<Record> writer =
- Avro.write(Files.localOutput(recordsFile))
- .schema(schema)
- .createWriterFunc(DataWriter::create)
- .build()) {
- writer.addAll(expectedRecords);
- }
-
- try (CloseableIterable<RowData> reader = createAvroReadBuilder(recordsFile, schema).build()) {
- Iterator<Record> expected = expectedRecords.iterator();
- Iterator<RowData> rows = reader.iterator();
- for (int i = 0; i < expectedRecords.size(); i++) {
- assertThat(rows).hasNext();
- TestHelpers.assertRowData(schema.asStruct(), flinkSchema, expected.next(), rows.next());
- }
- assertThat(rows).isExhausted();
- }
-
- File rowDataFile = File.createTempFile("junit", null, temp.toFile());
- assertThat(rowDataFile.delete()).isTrue();
-
- // Write the expected RowData into AVRO file, then read them into Record and assert with the
- // expected RowData list.
- try (FileAppender<RowData> writer =
- Avro.write(Files.localOutput(rowDataFile))
- .schema(schema)
- .createWriterFunc(ignore -> new FlinkAvroWriter(flinkSchema))
- .build()) {
- writer.addAll(expectedRows);
- }
-
- try (CloseableIterable<Record> reader =
- Avro.read(Files.localInput(rowDataFile))
- .project(schema)
- .createReaderFunc(DataReader::create)
- .build()) {
- Iterator<RowData> expected = expectedRows.iterator();
- Iterator<Record> records = reader.iterator();
- for (int i = 0; i < expectedRecords.size(); i += 1) {
- assertThat(records).hasNext();
- TestHelpers.assertRowData(schema.asStruct(), flinkSchema, records.next(), expected.next());
- }
- assertThat(records).isExhausted();
- }
- }
-
- private Record recordNumType(
- int id,
- int intV,
- float floatV,
- double doubleV,
- long date,
- long time,
- long timestamp,
- long bigint,
- double decimal) {
- Record record = GenericRecord.create(SCHEMA_NUM_TYPE);
- record.setField("id", id);
- record.setField("int", intV);
- record.setField("float", floatV);
- record.setField("double", doubleV);
- record.setField(
- "date", DateTimeUtil.dateFromDays((int) new Date(date).toLocalDate().toEpochDay()));
- record.setField("time", new Time(time).toLocalTime());
- record.setField("timestamp", DateTimeUtil.timestampFromMicros(timestamp * 1000));
- record.setField("bigint", bigint);
- record.setField("decimal", BigDecimal.valueOf(decimal));
- return record;
- }
-
- @Test
- public void testNumericTypes() throws IOException {
- List<Record> expected =
- ImmutableList.of(
- recordNumType(
- 2,
- Integer.MAX_VALUE,
- Float.MAX_VALUE,
- Double.MAX_VALUE,
- Long.MAX_VALUE,
- 1643811742000L,
- 1643811742000L,
- 1643811742000L,
- 10.24d),
- recordNumType(
- 2,
- Integer.MIN_VALUE,
- Float.MIN_VALUE,
- Double.MIN_VALUE,
- Long.MIN_VALUE,
- 1643811742000L,
- 1643811742000L,
- 1643811742000L,
- 10.24d));
-
- writeAndValidate(SCHEMA_NUM_TYPE, expected);
Review Comment:
This might be an interesting edge case to test in the `TestData`?
Min/max values for different columns? Could be especially interesting for timestamps/date etc
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.20: Support Avro and Parquet timestamp(9), unknown, and defaults [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue merged PR #12470:
URL: https://github.com/apache/iceberg/pull/12470
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.20: Support Avro and Parquet timestamp(9), unknown, and defaults [iceberg]
Posted by "danielcweeks (via GitHub)" <gi...@apache.org>.
danielcweeks commented on code in PR #12470:
URL: https://github.com/apache/iceberg/pull/12470#discussion_r1995967404
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetWriters.java:
##########
@@ -194,8 +177,104 @@ public ParquetValueWriter<?> primitive(LogicalType fType, PrimitiveType primitiv
}
}
- private static ParquetValueWriters.PrimitiveWriter<?> ints(
- LogicalType type, ColumnDescriptor desc) {
+ private static class LogicalTypeWriterBuilder
+ implements LogicalTypeAnnotationVisitor<ParquetValueWriter<?>> {
+ private final LogicalType flinkType;
+ private final ColumnDescriptor desc;
+
+ private LogicalTypeWriterBuilder(LogicalType flinkType, ColumnDescriptor desc) {
+ this.flinkType = flinkType;
+ this.desc = desc;
+ }
+
+ @Override
+ public Optional<ParquetValueWriter<?>> visit(StringLogicalTypeAnnotation strings) {
+ return Optional.of(strings(desc));
+ }
+
+ @Override
+ public Optional<ParquetValueWriter<?>> visit(EnumLogicalTypeAnnotation enums) {
+ return Optional.of(strings(desc));
+ }
+
+ @Override
+ public Optional<ParquetValueWriter<?>> visit(DecimalLogicalTypeAnnotation decimal) {
+ ParquetValueWriter<DecimalData> writer;
+ switch (desc.getPrimitiveType().getPrimitiveTypeName()) {
+ case INT32:
+ writer = decimalAsInteger(desc, decimal.getPrecision(), decimal.getScale());
+ break;
+ case INT64:
+ writer = decimalAsLong(desc, decimal.getPrecision(), decimal.getScale());
+ break;
+ case BINARY:
+ case FIXED_LEN_BYTE_ARRAY:
+ writer = decimalAsFixed(desc, decimal.getPrecision(), decimal.getScale());
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ "Unsupported base type for decimal: "
+ + desc.getPrimitiveType().getPrimitiveTypeName());
+ }
+ return Optional.of(writer);
+ }
+
+ @Override
+ public Optional<ParquetValueWriter<?>> visit(DateLogicalTypeAnnotation dates) {
+ return Optional.of(ints(flinkType, desc));
+ }
+
+ @Override
+ public Optional<ParquetValueWriter<?>> visit(TimeLogicalTypeAnnotation times) {
+ Preconditions.checkArgument(
+ LogicalTypeAnnotation.TimeUnit.MICROS.equals(times.getUnit()),
+ "Cannot write time in %s, only MICROS is supported",
+ times.getUnit());
+ return Optional.of(timeMicros(desc));
+ }
+
+ @Override
+ public Optional<ParquetValueWriter<?>> visit(TimestampLogicalTypeAnnotation timestamps) {
+ ParquetValueWriter<TimestampData> writer;
+ switch (timestamps.getUnit()) {
+ case NANOS:
+ writer = timestampNanos(desc);
+ break;
+ case MICROS:
+ writer = timestamps(desc);
+ break;
+ default:
+ throw new UnsupportedOperationException("Unsupported timestamp type: " + timestamps);
+ }
+
+ return Optional.of(writer);
+ }
+
+ @Override
+ public Optional<ParquetValueWriter<?>> visit(IntLogicalTypeAnnotation type) {
+ Preconditions.checkArgument(type.isSigned(), "Cannot write signed integer type: %s", type);
Review Comment:
Do you mean "cannot write _unsigned_ integer type: %s"?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support Avro timestamp(9), unknown, and defaults [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on PR #12470:
URL: https://github.com/apache/iceberg/pull/12470#issuecomment-2704918354
#12463 was merged and the changes for Parquet were small, so I included them here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.20: Support Avro and Parquet timestamp(9), unknown, and defaults [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #12470:
URL: https://github.com/apache/iceberg/pull/12470#discussion_r1996272927
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/data/AbstractTestFlinkAvroReaderWriter.java:
##########
@@ -1,182 +0,0 @@
-/*
- * 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.flink.data;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-import java.io.File;
-import java.io.IOException;
-import java.math.BigDecimal;
-import java.sql.Date;
-import java.sql.Time;
-import java.util.Iterator;
-import java.util.List;
-import org.apache.flink.table.data.RowData;
-import org.apache.flink.table.types.logical.RowType;
-import org.apache.iceberg.Files;
-import org.apache.iceberg.Schema;
-import org.apache.iceberg.avro.Avro;
-import org.apache.iceberg.data.DataTest;
-import org.apache.iceberg.data.GenericRecord;
-import org.apache.iceberg.data.RandomGenericData;
-import org.apache.iceberg.data.Record;
-import org.apache.iceberg.data.avro.DataReader;
-import org.apache.iceberg.data.avro.DataWriter;
-import org.apache.iceberg.flink.FlinkSchemaUtil;
-import org.apache.iceberg.flink.TestHelpers;
-import org.apache.iceberg.io.CloseableIterable;
-import org.apache.iceberg.io.FileAppender;
-import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
-import org.apache.iceberg.relocated.com.google.common.collect.Lists;
-import org.apache.iceberg.types.Types;
-import org.apache.iceberg.util.DateTimeUtil;
-import org.junit.jupiter.api.Test;
-
-public abstract class AbstractTestFlinkAvroReaderWriter extends DataTest {
-
- private static final int NUM_RECORDS = 100;
-
- private static final Schema SCHEMA_NUM_TYPE =
- new Schema(
- Types.NestedField.optional(1, "id", Types.IntegerType.get()),
- Types.NestedField.optional(2, "int", Types.IntegerType.get()),
- Types.NestedField.optional(3, "float", Types.FloatType.get()),
- Types.NestedField.optional(4, "double", Types.DoubleType.get()),
- Types.NestedField.optional(5, "date", Types.DateType.get()),
- Types.NestedField.optional(6, "time", Types.TimeType.get()),
- Types.NestedField.optional(7, "timestamp", Types.TimestampType.withoutZone()),
- Types.NestedField.optional(8, "bigint", Types.LongType.get()),
- Types.NestedField.optional(9, "decimal", Types.DecimalType.of(4, 2)));
-
- @Override
- protected void writeAndValidate(Schema schema) throws IOException {
- List<Record> expectedRecords = RandomGenericData.generate(schema, NUM_RECORDS, 1991L);
- writeAndValidate(schema, expectedRecords);
- }
-
- protected abstract Avro.ReadBuilder createAvroReadBuilder(File recordsFile, Schema schema);
-
- @Override
- protected void writeAndValidate(Schema schema, List<Record> expectedRecords) throws IOException {
- RowType flinkSchema = FlinkSchemaUtil.convert(schema);
- List<RowData> expectedRows = Lists.newArrayList(RandomRowData.convert(schema, expectedRecords));
-
- File recordsFile = File.createTempFile("junit", null, temp.toFile());
- assertThat(recordsFile.delete()).isTrue();
-
- // Write the expected records into AVRO file, then read them into RowData and assert with the
- // expected Record list.
- try (FileAppender<Record> writer =
- Avro.write(Files.localOutput(recordsFile))
- .schema(schema)
- .createWriterFunc(DataWriter::create)
- .build()) {
- writer.addAll(expectedRecords);
- }
-
- try (CloseableIterable<RowData> reader = createAvroReadBuilder(recordsFile, schema).build()) {
- Iterator<Record> expected = expectedRecords.iterator();
- Iterator<RowData> rows = reader.iterator();
- for (int i = 0; i < expectedRecords.size(); i++) {
- assertThat(rows).hasNext();
- TestHelpers.assertRowData(schema.asStruct(), flinkSchema, expected.next(), rows.next());
- }
- assertThat(rows).isExhausted();
- }
-
- File rowDataFile = File.createTempFile("junit", null, temp.toFile());
- assertThat(rowDataFile.delete()).isTrue();
-
- // Write the expected RowData into AVRO file, then read them into Record and assert with the
- // expected RowData list.
- try (FileAppender<RowData> writer =
- Avro.write(Files.localOutput(rowDataFile))
- .schema(schema)
- .createWriterFunc(ignore -> new FlinkAvroWriter(flinkSchema))
- .build()) {
- writer.addAll(expectedRows);
- }
-
- try (CloseableIterable<Record> reader =
- Avro.read(Files.localInput(rowDataFile))
- .project(schema)
- .createReaderFunc(DataReader::create)
- .build()) {
- Iterator<RowData> expected = expectedRows.iterator();
- Iterator<Record> records = reader.iterator();
- for (int i = 0; i < expectedRecords.size(); i += 1) {
- assertThat(records).hasNext();
- TestHelpers.assertRowData(schema.asStruct(), flinkSchema, records.next(), expected.next());
- }
- assertThat(records).isExhausted();
- }
- }
-
- private Record recordNumType(
- int id,
- int intV,
- float floatV,
- double doubleV,
- long date,
- long time,
- long timestamp,
- long bigint,
- double decimal) {
- Record record = GenericRecord.create(SCHEMA_NUM_TYPE);
- record.setField("id", id);
- record.setField("int", intV);
- record.setField("float", floatV);
- record.setField("double", doubleV);
- record.setField(
- "date", DateTimeUtil.dateFromDays((int) new Date(date).toLocalDate().toEpochDay()));
- record.setField("time", new Time(time).toLocalTime());
- record.setField("timestamp", DateTimeUtil.timestampFromMicros(timestamp * 1000));
- record.setField("bigint", bigint);
- record.setField("decimal", BigDecimal.valueOf(decimal));
- return record;
- }
-
- @Test
- public void testNumericTypes() throws IOException {
- List<Record> expected =
- ImmutableList.of(
- recordNumType(
- 2,
- Integer.MAX_VALUE,
- Float.MAX_VALUE,
- Double.MAX_VALUE,
- Long.MAX_VALUE,
- 1643811742000L,
- 1643811742000L,
- 1643811742000L,
- 10.24d),
- recordNumType(
- 2,
- Integer.MIN_VALUE,
- Float.MIN_VALUE,
- Double.MIN_VALUE,
- Long.MIN_VALUE,
- 1643811742000L,
- 1643811742000L,
- 1643811742000L,
- 10.24d));
-
- writeAndValidate(SCHEMA_NUM_TYPE, expected);
Review Comment:
These cases are tested in the random data generator, so this is duplication.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org